Update NailGun. Summary: Update the version of NailGun that buckd uses to https://github.com/martylamb/nailgun/commits/fcc24ecd63ead63f160e96f6a0640f6b4280f72d, which uses a FixedThreadPool instead of CachedThreadPool to reduce the number of threads that Buck uses and notifies blocked threads on client disconnection so that threads aren't leaked. Test Plan: 0) buck test --all 1) buckd 2) buck build buck 3) buck build buck 4) check that buckd is using <70 threads
diff --git a/lib/README.txt b/lib/README.txt index 53963da..38f1962 100644 --- a/lib/README.txt +++ b/lib/README.txt
@@ -9,15 +9,14 @@ nailgun-server-0.9.2-SNAPSHOT.jar and nailgun-server-0.9.2-SNAPSHOT-sources.jar were -built from a fork of Nailgun which adds support for detecting client disconnection at -https://github.com/jimpurbrick/nailgun.git +built from https://github.com/martylamb/nailgun.git To regenerate these jars: 0) install maven (brew install maven) - 1) git clone https://github.com/jimpurbrick/nailgun.git + 1) git clone https://github.com/martylamb/nailgun 2) cd nailgun - 3) git checkout d005c16f13d42489ac1ab428b15c3d9cdb1ada31 + 3) git checkout fcc24ecd63ead63f160e96f6a0640f6b4280f72d 4) mvn clean install 5) copy nailgun-server/target/nailgun-server-0.9.2-SNAPSHOT.jar and nailgun-server/target/nailgun-server-0.9.2-SNAPSHOT-sources.jar to your buck/lib directory
diff --git a/lib/nailgun-server-0.9.2-SNAPSHOT-sources.jar b/lib/nailgun-server-0.9.2-SNAPSHOT-sources.jar index a595667..4513f48 100644 --- a/lib/nailgun-server-0.9.2-SNAPSHOT-sources.jar +++ b/lib/nailgun-server-0.9.2-SNAPSHOT-sources.jar Binary files differ
diff --git a/lib/nailgun-server-0.9.2-SNAPSHOT.jar b/lib/nailgun-server-0.9.2-SNAPSHOT.jar index 8d45dda..308dd9f 100644 --- a/lib/nailgun-server-0.9.2-SNAPSHOT.jar +++ b/lib/nailgun-server-0.9.2-SNAPSHOT.jar Binary files differ
diff --git a/third-party/nailgun/README.md b/third-party/nailgun/README.md index 213d4f4..df471ba 100644 --- a/third-party/nailgun/README.md +++ b/third-party/nailgun/README.md
@@ -16,6 +16,5 @@ For more information, see [the nailgun website](http://martiansoftware.com/nailgun/). -Buck currently uses a fork of nailgun hosted at https://github.com/jimpurbrick/nailgun. -This fork adds support for interrupting server processing when client disconnection is -detected, changes which are currently being accepted in to the main nailgun repository. \ No newline at end of file +Buck currently uses https://github.com/martylamb/nailgun at +fcc24ecd63ead63f160e96f6a0640f6b4280f72d
diff --git a/third-party/nailgun/nailgun-client/ng.c b/third-party/nailgun/nailgun-client/ng.c index bd7a390..059b05f 100644 --- a/third-party/nailgun/nailgun-client/ng.c +++ b/third-party/nailgun/nailgun-client/ng.c
@@ -27,7 +27,7 @@ #include <netdb.h> #include <netinet/in.h> #include <sys/socket.h> - #include <sys/time.h> + #include <sys/time.h> #include <sys/types.h> #endif @@ -96,6 +96,8 @@ #define CHUNKTYPE_SENDINPUT 'S' #define CHUNKTYPE_HEARTBEAT 'H' +#define HEARTBEAT_TIMEOUT_MILLIS 500 + /* the following is required to compile for hp-ux originally posted at http://jira.codehaus.org/browse/JRUBY-2346 @@ -111,7 +113,12 @@ char buf[BUFSIZE]; /* track whether server is ready to receive */ -int readyToSend = 0; +#ifdef WIN32 + HANDLE readyToSend = 0; + HANDLE sending = 0; +#else + int readyToSend = 0; +#endif /** * Clean up the application. @@ -192,12 +199,13 @@ } /** - * Sends a chunk header noting the specified payload size and chunk type. + * Sends a chunk noting the specified payload size and chunk type. + * Waits for sending mutex on Win32. * * @param size the payload size * @param chunkType the chunk type identifier */ -void sendHeader(unsigned int size, char chunkType) { +void sendChunk(unsigned int size, char chunkType, char* buf) { /* buffer used for reading and writing chunk headers */ char header[CHUNK_HEADER_LEN]; @@ -207,7 +215,20 @@ header[3] = size & 0xff; header[4] = chunkType; +#ifdef WIN32 + if (WaitForSingleObject(sending, INFINITE) != WAIT_OBJECT_0) { + handleError(); + } +#endif + sendAll(nailgunsocket, header, CHUNK_HEADER_LEN); + if (size > 0) { + sendAll(nailgunsocket, buf, size); + } + +#ifdef WIN32 + ReleaseMutex(sending); +#endif } /** @@ -228,15 +249,14 @@ i = read(f, buf, BUFSIZE); while (i > 0) { - sendHeader(i, CHUNKTYPE_LONGARG); - sendAll(nailgunsocket, buf, i); + sendChunk(i, CHUNKTYPE_LONGARG, buf); i = read(f, buf, BUFSIZE); } if (i < 0) { perror("--nailgun-filearg"); return 1; } - sendHeader(0, CHUNKTYPE_LONGARG); + sendChunk(0, CHUNKTYPE_LONGARG, buf); close(f); return 0; @@ -250,8 +270,7 @@ */ void sendText(char chunkType, char *text) { int len = text ? strlen(text) : 0; - sendHeader(len, chunkType); - sendAll(nailgunsocket, text, len); + sendChunk(len, chunkType, text); } /** @@ -278,9 +297,9 @@ int thisPass = 0; thisPass = recv(nailgunsocket, buf, bytesToRead, MSG_WAITALL); - if (thisPass < bytesToRead) handleSocketClose(); - - + if (thisPass == 0) { + handleSocketClose(); + } bytesRead += thisPass; bytesCopied = 0; @@ -304,6 +323,17 @@ } } +unsigned long recvToBuffer(unsigned long len) { + unsigned long bytesRead = 0; + while(bytesRead < len) { + int thisPass = recv(nailgunsocket, buf + bytesRead, len - bytesRead, MSG_WAITALL); + if (thisPass == 0) { + handleSocketClose(); + } + bytesRead += thisPass; + } + return bytesRead; +} /** * Processes an exit chunk from the server. This is just a string @@ -315,7 +345,7 @@ void processExit(char *buf, unsigned long len) { int exitcode; int bytesToRead = (BUFSIZE - 1 < len) ? BUFSIZE - 1 : len; - int bytesRead = recv(nailgunsocket, buf, bytesToRead, MSG_WAITALL); + int bytesRead = recvToBuffer(bytesToRead); if (bytesRead < 0) { handleSocketClose(); @@ -336,26 +366,44 @@ * @param len the number of bytes to send */ void sendStdin(char *buf, unsigned int len) { +#ifndef WIN32 readyToSend = 0; - sendHeader(len, CHUNKTYPE_STDIN); - sendAll(nailgunsocket, buf, len); +#endif + sendChunk(len, CHUNKTYPE_STDIN, buf); } /** * Sends a stdin-eof chunk to the nailgun server */ void processEof() { - sendHeader(0, CHUNKTYPE_STDIN_EOF); + sendChunk(0, CHUNKTYPE_STDIN_EOF, buf); } /** * Sends a heartbeat chunk to let the server know the client is still alive. */ void sendHeartbeat() { - sendHeader(0, CHUNKTYPE_HEARTBEAT); + sendChunk(0, CHUNKTYPE_HEARTBEAT, buf); } #ifdef WIN32 + +HANDLE createEvent(BOOL manualReset) { + return CreateEvent(NULL, /* default security */ + manualReset, + FALSE, /* initial state unsignalled */ + NULL /* unnamed event */); +} + +DWORD WINAPI sendHeartbeats(LPVOID lpParameter) { + + /* this could be made more efficient by only sending heartbeats when stdin chunks aren't being sent */ + for (;;) { + Sleep(HEARTBEAT_TIMEOUT_MILLIS); + sendHeartbeat(); + } +} + /** * Thread main for reading from stdin and sending */ @@ -363,23 +411,32 @@ /* buffer used for reading and sending stdin chunks */ char wbuf[BUFSIZE]; - for (;;) { - DWORD numberOfBytes = 0; + /* number of bytes read */ + DWORD numberOfBytes; - if (readyToSend && !ReadFile(NG_STDIN_FILENO, wbuf, BUFSIZE, &numberOfBytes, NULL)) { - if (numberOfBytes != 0) { - handleError(); - } + for (;;) { + + /* wait for ready to send */ + if(WaitForSingleObject(readyToSend, INFINITE) != WAIT_OBJECT_0) { + handleError(); } + /* read data from stdin */ + if (! ReadFile(NG_STDIN_FILENO, wbuf, BUFSIZE, &numberOfBytes, NULL)) { + if (numberOfBytes != 0) { + handleError(); + } + } + + /* send data to server */ if (numberOfBytes > 0) { sendStdin(wbuf, numberOfBytes); } else { processEof(); break; } - } + } return 0; } #else @@ -408,6 +465,10 @@ WSADATA win_socket_data; /* required to initialise winsock */ WSAStartup(2, &win_socket_data); + + /* create flow control event and mutex */ + readyToSend = createEvent(FALSE); + sending = CreateMutex(NULL, FALSE, NULL); } #endif @@ -440,6 +501,10 @@ if (!CreateThread(&securityAttributes, 0, &processStdin, NULL, 0, &threadId)) { handleError(); } + + if (!CreateThread(&securityAttributes, 0, &sendHeartbeats, NULL, 0, &threadId)) { + handleError(); + } } #endif @@ -449,16 +514,11 @@ void processnailgunstream() { /*for (;;) {*/ - int bytesRead = 0; unsigned long len; char chunkType; - bytesRead = recv(nailgunsocket, buf, CHUNK_HEADER_LEN, MSG_WAITALL); + recvToBuffer(CHUNK_HEADER_LEN); - if (bytesRead < CHUNK_HEADER_LEN) { - handleSocketClose(); - } - len = ((buf[0] << 24) & 0xff000000) | ((buf[1] << 16) & 0x00ff0000) | ((buf[2] << 8) & 0x0000ff00) @@ -474,7 +534,11 @@ case CHUNKTYPE_EXIT: processExit(buf, len); break; case CHUNKTYPE_SENDINPUT: - readyToSend = 1; +#ifdef WIN32 + SetEvent(readyToSend); +#else + readyToSend = 1; +#endif break; default: fprintf(stderr, "Unexpected chunk type %d ('%c')\n", chunkType, chunkType); cleanUpAndExit(NAILGUN_UNEXPECTED_CHUNKTYPE); @@ -556,7 +620,6 @@ fd_set readfds; int eof = 0; struct timeval readtimeout; - #endif #ifdef WIN32 @@ -694,6 +757,7 @@ /* initialise the std-* handles and the thread to send stdin to the server */ #ifdef WIN32 initIo(); + winStartInput(); #endif /* stream forwarding loop */ @@ -709,22 +773,22 @@ FD_SET(nailgunsocket, &readfds); memset(&readtimeout, '\0', sizeof(readtimeout)); - readtimeout.tv_usec = 100000; + readtimeout.tv_usec = HEARTBEAT_TIMEOUT_MILLIS * 1000; if(select (nailgunsocket + 1, &readfds, NULL, NULL, &readtimeout) == -1) { perror("select"); } if (FD_ISSET(nailgunsocket, &readfds)) { #endif - processnailgunstream(); + processnailgunstream(); #ifndef WIN32 } else if (FD_ISSET(NG_STDIN_FILENO, &readfds)) { - if (!processStdin()) { - FD_CLR(NG_STDIN_FILENO, &readfds); - eof = 1; - } + if (!processStdin()) { + FD_CLR(NG_STDIN_FILENO, &readfds); + eof = 1; + } } else { - sendHeartbeat(); + sendHeartbeat(); } #endif }