Author: mturk
Date: Sat Jul 9 12:09:51 2011
New Revision: 1144660
URL: http://svn.apache.org/viewvc?rev=1144660&view=rev
Log:
Fix win32 select selector. We need a copy before select
Modified:
commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
commons/sandbox/runtime/trunk/src/main/native/os/win32/selectset.c
commons/sandbox/runtime/trunk/src/main/native/os/win32/util.c
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
Modified: commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c?rev=1144660&r1=1144659&r2=1144660&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c
(original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/win32/localsock.c Sat Jul
9 12:09:51 2011
@@ -78,7 +78,8 @@ ACR_NET_EXPORT(jlong, LocalDescriptor, s
{
SOCKET sd;
int rc = 0;
- int type = ACR_DT_LSOCK;
+ int type = 0;
+ int desc = ACR_DT_LSOCK;
wls_fd_t *sp;
switch (stype) {
@@ -102,7 +103,7 @@ ACR_NET_EXPORT(jlong, LocalDescriptor, s
rc = ACR_GET_NETOS_ERROR();
closesocket(sd);
}
- type |= ACR_DT_NBLOCK;
+ desc |= ACR_DT_NBLOCK;
}
if (rc != 0) {
ACR_THROW_NET_ERROR(rc);
@@ -112,7 +113,7 @@ ACR_NET_EXPORT(jlong, LocalDescriptor, s
closesocket(sd);
return 0;
}
- sp->fd.type = type;
+ sp->fd.type = desc;
sp->fd.refs = 1;
sp->fd.u.s = sd;
sp->fh = 0;
@@ -136,7 +137,7 @@ ACR_NET_EXPORT(jint, LocalDescriptor, bl
ACR_NET_EXPORT(jboolean, LocalDescriptor, isBlocking0)(JNI_STDARGS, jlong fp)
{
acr_fd_t *fd = J2P(fp, acr_fd_t *);
- return fd->type & ACR_DT_NBLOCK ? JNI_FALSE : JNI_TRUE;
+ return (fd->type & ACR_DT_NBLOCK) == ACR_DT_NBLOCK ? JNI_FALSE : JNI_TRUE;
}
ACR_NET_EXPORT(jint, LocalEndpoint, connect0)(JNI_STDARGS, jlong fp,
@@ -249,7 +250,6 @@ ACR_NET_EXPORT(jint, LocalServerEndpoint
rc = ACR_GET_NETOS_ERROR();
goto failed;
}
-
SOCKADDR_RELEASE(ba, aa);
wd->fh = sfh;
return 0;
@@ -282,6 +282,7 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
acr_sockaddr_t *aa;
struct sockaddr_in sa;
int sas = ISIZEOF(sa);
+ int type;
wls_fd_t *sp;
wls_fd_t *wd = J2P(fp, wls_fd_t *);
@@ -292,13 +293,15 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
ACR_THROW_NET_ERRNO();
return 0;
}
- if (block == JNI_FALSE) {
+ type = wd->fd.type;
+ if (block == JNI_FALSE && (type & ACR_DT_NBLOCK) == 0) {
int rc = AcrNonblock(sd, 1);
if (rc != 0) {
closesocket(sd);
ACR_THROW_NET_ERROR(rc);
return 0;
}
+ type |= ACR_DT_NBLOCK;
}
if ((sp = ACR_TALLOC(wls_fd_t)) == 0) {
closesocket(sd);
@@ -311,7 +314,7 @@ ACR_NET_EXPORT(jlong, LocalServerEndpoin
aa->port = ntohs(sa.sin_port);
}
SOCKADDR_RELEASE(ba, aa);
- sp->fd.type = ACR_DT_LSOCK;
+ sp->fd.type = type;
sp->fd.refs = 1;
sp->fd.u.s = sd;
return P2J(sp);
Modified: commons/sandbox/runtime/trunk/src/main/native/os/win32/selectset.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/win32/selectset.c?rev=1144660&r1=1144659&r2=1144660&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/win32/selectset.c
(original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/win32/selectset.c Sat Jul
9 12:09:51 2011
@@ -307,7 +307,7 @@ ACR_NET_EXPORT(jint, LegacySelector, add
if (events & ACR_OP_OUT) {
FD_SET(fd->u.s, &ps->wrset);
}
- if (events & ~(ACR_OP_INP | ACR_OP_OUT)) {
+ if (events & 0xFFF0) {
FD_SET(fd->u.s, &ps->exset);
}
ps->used++;
@@ -368,6 +368,8 @@ ACR_NET_EXPORT(jint, LegacySelector, wai
struct timeval tv;
struct timeval *tp = 0;
jshort *pevents;
+ fd_set rdset, wrset, exset;
+
acr_pollset_t *ps = J2P(pollset, acr_pollset_t *);
EnterCriticalSection(&ps->mutex);
@@ -387,17 +389,20 @@ ACR_NET_EXPORT(jint, LegacySelector, wai
LeaveCriticalSection(&ps->mutex);
return 0;
}
+ memcpy(&rdset, &ps->rdset, sizeof(fd_set));
+ memcpy(&wrset, &ps->wrset, sizeof(fd_set));
+ memcpy(&exset, &ps->exset, sizeof(fd_set));
AcrAtomic32Set(&ps->state, PSS_POLL);
LeaveCriticalSection(&ps->mutex);
- if (timeout > 0) {
+ if (timeout >= 0) {
tp = &tv;
tp->tv_sec = (long)(timeout / 1000);
tp->tv_usec = (long)(timeout % 1000) * 1000;
}
- ns = select(ps->used, &ps->rdset, &ps->wrset, &ps->exset, tp);
+ ns = select(ps->used, &rdset, &wrset, &exset, tp);
if (ns == SOCKET_ERROR)
- rc = ACR_GET_OS_ERROR();
+ rc = ACR_GET_NETOS_ERROR();
EnterCriticalSection(&ps->mutex);
if (AcrAtomic32Equ(&ps->state, PSS_DESTROY)) {
/* Interrupted by destroy0 */
@@ -407,7 +412,7 @@ ACR_NET_EXPORT(jint, LegacySelector, wai
return 0;
}
if (rc != 0) {
- /* Error during poll */
+ /* Error during select */
if (rc == WSAENOTSOCK) {
/* Find the closed socket(s) */
int optval;
@@ -418,11 +423,14 @@ ACR_NET_EXPORT(jint, LegacySelector, wai
for (i = 1; i < ps->used; i++) {
ps->ooset[i].revents = 0;
if (getsockopt(ps->ooset[i].sd, SOL_SOCKET, SO_TYPE, (char
*)&optval, &optlen) == -1) {
- if (WSAGetLastError() == WSAENOTSOCK) {
- ps->ooset[i].revents = ACR_OP_NVAL;
- pevents[i] = ACR_OP_ERROR | ACR_OP_NVAL;
- (*env)->SetObjectArrayElement(env, rs, rv++,
ps->ooset[i].obj);
- }
+ ps->ooset[i].revents = ACR_OP_NVAL;
+ pevents[i] = ACR_OP_ERROR | ACR_OP_NVAL;
+ (*env)->SetObjectArrayElement(env, rs, rv++,
ps->ooset[i].obj);
+ }
+ else if (optval != 0) {
+ ps->ooset[i].revents = ACR_OP_NVAL;
+ pevents[i] = ACR_OP_ERROR | ACR_OP_NVAL;
+ (*env)->SetObjectArrayElement(env, rs, rv++,
ps->ooset[i].obj);
}
}
if (rv != 0)
@@ -477,11 +485,11 @@ ACR_NET_EXPORT(jint, LegacySelector, wai
for (i = 0; i < ps->used; i++) {
/* Get selected events */
ps->ooset[i].revents = 0;
- if (FD_ISSET(ps->ooset[i].sd, &ps->rdset))
+ if (FD_ISSET(ps->ooset[i].sd, &rdset))
ps->ooset[i].revents |= ACR_OP_INP;
- if (FD_ISSET(ps->ooset[i].sd, &ps->wrset))
+ if (FD_ISSET(ps->ooset[i].sd, &wrset))
ps->ooset[i].revents |= ACR_OP_OUT;
- if (FD_ISSET(ps->ooset[i].sd, &ps->exset))
+ if (FD_ISSET(ps->ooset[i].sd, &exset))
ps->ooset[i].revents |= ACR_OP_ERROR;
if (ps->ooset[i].revents != 0) {
if (i == 0) {
@@ -534,6 +542,7 @@ cleanup:
FD_CLR(ps->ooset[i].sd, &ps->rdset);
FD_CLR(ps->ooset[i].sd, &ps->wrset);
FD_CLR(ps->ooset[i].sd, &ps->exset);
+
for (++i; i < used; i++) {
if (ps->ooset[i].revents != 0) {
/* Skip signaled descriptor */
Modified: commons/sandbox/runtime/trunk/src/main/native/os/win32/util.c
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/native/os/win32/util.c?rev=1144660&r1=1144659&r2=1144660&view=diff
==============================================================================
--- commons/sandbox/runtime/trunk/src/main/native/os/win32/util.c (original)
+++ commons/sandbox/runtime/trunk/src/main/native/os/win32/util.c Sat Jul 9
12:09:51 2011
@@ -544,12 +544,12 @@ AcrReadFileDataA(const char *name, char
if (MultiByteToWideChar(CP_UTF8, 0, name, -1, wname, ACR_PATH_MAX) == 0)
return 0;
fh = CreateFileW(wname, GENERIC_READ,
- FILE_SHARE_READ | FILE_SHARE_WRITE,
+ FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
GetSaWithNullDacl(0, JNI_FALSE), OPEN_EXISTING, 0, 0);
if (IS_INVALID_HANDLE(fh))
return 0;
if (sbuf != 0 && *len > 0) {
- if (!ReadFile(fh, sbuf, *(len - 1), &bpos, 0)) {
+ if (!ReadFile(fh, sbuf, *len - 1, &bpos, 0)) {
rc = GetLastError();
goto failed;
}
Modified:
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
URL:
http://svn.apache.org/viewvc/commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java?rev=1144660&r1=1144659&r2=1144660&view=diff
==============================================================================
---
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
(original)
+++
commons/sandbox/runtime/trunk/src/main/test/org/apache/commons/runtime/TestPosixEndpoint.java
Sat Jul 9 12:09:51 2011
@@ -28,7 +28,7 @@ public class TestPosixEndpoint extends A
{
@BeforeSuite
- @Test(groups = { "posix" })
+ @Test(groups = { "core" })
public void setUp()
throws Exception
{
@@ -36,7 +36,7 @@ public class TestPosixEndpoint extends A
File f = new File("/tmp/acrposixep1.sock");
f.delete();
}
-
+
private Object sync = new Object();
class Acceptor extends Thread
@@ -59,21 +59,23 @@ public class TestPosixEndpoint extends A
assertEquals(ps.size(), 1);
} catch (Exception x) {
fail("Acceptor setup failed " + x.toString());
+ running = false;
+ return;
}
while (true) {
try {
- System.out.println("Accepting connection ...");
synchronized (sync) {
// Notify that we are ready to
// accept the connections
sync.notifyAll();
}
+ System.out.print("Accepting connection ... ");
List<SelectionKey> set = ps.select();
if (set.size() == 0) {
- System.out.println("Interrupted.");
+ System.out.println("interrupted.");
break;
}
- System.out.println("Accepted " + set.size() + "
connection");
+ System.out.println("done (" + set.size() + " connection)");
assertEquals(set.size(), 1);
sk = set.get(0);
LocalEndpoint e =
((LocalServerEndpoint)sk.endpoint()).accept();
@@ -83,6 +85,7 @@ public class TestPosixEndpoint extends A
assertEquals(set.size(), 0);
} catch (Exception x) {
fail("Accept failed " + x.toString());
+ break;
}
}
try {
@@ -90,14 +93,15 @@ public class TestPosixEndpoint extends A
} catch (Exception x) {
fail("Acceptor shutdown failed " + x.toString());
}
+ running = false;
}
}
- @Test(groups = { "posix" })
+ @Test(groups = { "core" })
public void connectLocalEndpoint()
throws Exception
{
-
+
Selector ps = Selector.open(EndpointType.LOCAL);
LocalServerEndpoint ss = new LocalServerEndpoint();
LocalEndpointAddress sa = new
LocalEndpointAddress("/tmp/acrposixep1.sock");
@@ -134,17 +138,21 @@ public class TestPosixEndpoint extends A
cs.connect(sa);
assertTrue(cs.isBlocking());
cs.close();
+ ps.interrupt();
try {
synchronized (sync) {
// Wait until Acceptor process the accepted connection
//
- sync.wait();
+ if (aw.running)
+ sync.wait();
}
Thread.sleep(200);
} catch (InterruptedException x) {
// Ignore
- }
+ }
+ System.out.print("Interrupting selector ... ");
ps.interrupt();
+ System.out.println("OK");
// Wait for the Acceptor thread to finish
aw.join();
List<SelectionKey> set = ps.clear();
