{"thread":{"id":"2034","subject":"[PATCH] Add support for parallel HTTP transfers","startedAt":"2005-10-06T18:54:12Z","lastAt":"2005-10-06T18:54:12Z","messageCount":1,"participants":["Nick Hengeveld"],"isPatch":true,"patchVersion":1,"patchTotal":null},"messages":[{"id":"9771","messageId":"20051006185412.GG15593@reactrix.com","threadId":"2034","inReplyTo":null,"subject":"[PATCH] Add support for parallel HTTP transfers","fromName":"Nick Hengeveld","fromEmail":"nickh@reactrix.com","sentAt":"2005-10-06T18:54:12Z","receivedAt":"2005-10-06T18:54:12Z","isPatch":true,"sender":{"key":"nickh@reactrix.com","avatar":null},"body":"Add support for parallel HTTP transfers.  Prefetch populates a queue of\nobjects to transfer and starts feeding requests to an active request\nqueue for processing; fetch_object keeps the active queue moving\nwhile the specified object is being transferred.  The size of the active\nqueue can be restricted using -r and defaults to 5 concurrent transfers.\nRequests for objects that are not prefetched are also processed via the\nactive queue.\n\nSigned-off-by: Nick Hengeveld <nickh@reactrix.com>\n\n\n---\n\nThis patch replaces the previous version.  It seems that the instability\nI experienced with empty server responses was related to request timeouts\nbecause the active queue wasn't being processed during a non-object fetch.\n\n\n http-fetch.c |  810 ++++++++++++++++++++++++++++++++++++++++++----------------\n 1 files changed, 587 insertions(+), 223 deletions(-)\n\n8dcb9f1bdfa5d4229ef031f8d01cc8cde0d54bbc\ndiff --git a/http-fetch.c b/http-fetch.c\n--- a/http-fetch.c\n+++ b/http-fetch.c\n@@ -6,6 +6,8 @@\n #include <curl/curl.h>\n #include <curl/easy.h>\n \n+#define DEFAULT_MAX_REQUESTS 5\n+\n #if LIBCURL_VERSION_NUM < 0x070704\n #define curl_global_cleanup() do { /* nothing */ } while(0)\n #endif\n@@ -16,13 +18,16 @@\n #define PREV_BUF_SIZE 4096\n #define RANGE_HEADER_SIZE 30\n \n-static CURL *curl;\n+static int max_requests = DEFAULT_MAX_REQUESTS;\n+static int active_requests = 0;\n+static int data_received;\n+\n+static CURLM *curlm;\n+static CURL *curl_default;\n static struct curl_slist *no_pragma_header;\n static struct curl_slist *no_range_header;\n static char curl_errorstr[CURL_ERROR_SIZE];\n \n-static char *initial_base;\n-\n struct alt_base\n {\n \tchar *base;\n@@ -33,11 +38,46 @@ struct alt_base\n \n static struct alt_base *alt = NULL;\n \n-static SHA_CTX c;\n-static z_stream stream;\n+enum transfer_state {\n+\tWAITING,\n+\tABORTED,\n+\tACTIVE,\n+\tCOMPLETE,\n+};\n+\n+struct transfer_request\n+{\n+\tunsigned char sha1[20];\n+\tstruct alt_base *repo;\n+\tchar *url;\n+\tchar filename[PATH_MAX];\n+\tchar tmpfile[PATH_MAX];\n+\tint local;\n+\tenum transfer_state state;\n+\tCURLcode curl_result;\n+\tchar errorstr[CURL_ERROR_SIZE];\n+\tlong http_code;\n+\tunsigned char real_sha1[20];\n+\tSHA_CTX c;\n+\tz_stream stream;\n+\tint zret;\n+\tint rename;\n+\tstruct active_request_slot *slot;\n+\tstruct transfer_request *next;\n+};\n+\n+struct active_request_slot\n+{\n+\tCURL *curl;\n+\tFILE *local;\n+\tint in_use;\n+\tint done;\n+\tCURLcode curl_result;\n+\tstruct active_request_slot *next;\n+};\n \n-static int local;\n-static int zret;\n+static struct transfer_request *request_queue_head = NULL;\n+static struct active_request_slot *active_queue_head = NULL;\n \n static int curl_ssl_verify;\n static char *ssl_cert;\n@@ -60,6 +100,7 @@ static size_t fwrite_buffer(void *ptr, s\n                 size = buffer->size - buffer->posn;\n         memcpy(buffer->buffer + buffer->posn, ptr, size);\n         buffer->posn += size;\n+\tdata_received++;\n         return size;\n }\n \n@@ -69,28 +110,28 @@ static size_t fwrite_sha1_file(void *ptr\n \tunsigned char expn[4096];\n \tsize_t size = eltsize * nmemb;\n \tint posn = 0;\n+\tstruct transfer_request *request = (struct transfer_request *)data;\n \tdo {\n-\t\tssize_t retval = write(local, ptr + posn, size - posn);\n+\t\tssize_t retval = write(request->local,\n+\t\t\t\t       ptr + posn, size - posn);\n \t\tif (retval < 0)\n \t\t\treturn posn;\n \t\tposn += retval;\n \t} while (posn < size);\n \n-\tstream.avail_in = size;\n-\tstream.next_in = ptr;\n+\trequest->stream.avail_in = size;\n+\trequest->stream.next_in = ptr;\n \tdo {\n-\t\tstream.next_out = expn;\n-\t\tstream.avail_out = sizeof(expn);\n-\t\tzret = inflate(&stream, Z_SYNC_FLUSH);\n-\t\tSHA1_Update(&c, expn, sizeof(expn) - stream.avail_out);\n-\t} while (stream.avail_in && zret == Z_OK);\n+\t\trequest->stream.next_out = expn;\n+\t\trequest->stream.avail_out = sizeof(expn);\n+\t\trequest->zret = inflate(&request->stream, Z_SYNC_FLUSH);\n+\t\tSHA1_Update(&request->c, expn,\n+\t\t\t    sizeof(expn) - request->stream.avail_out);\n+\t} while (request->stream.avail_in && request->zret == Z_OK);\n+\tdata_received++;\n \treturn size;\n }\n \n-void prefetch(unsigned char *sha1)\n-{\n-}\n-\n int relink_or_rename(char *old, char *new) {\n \tint ret;\n \n@@ -110,10 +151,369 @@ int relink_or_rename(char *old, char *ne\n \treturn 0;\n }\n \n+void process_curl_messages();\n+void process_request_queue();\n+\n+struct active_request_slot *get_active_slot()\n+{\n+\tstruct active_request_slot *slot = active_queue_head;\n+\tstruct active_request_slot *newslot;\n+\tint num_transfers;\n+\n+\t/* Wait for a slot to open up if the queue is full */\n+\twhile (active_requests >= max_requests) {\n+\t\tcurl_multi_perform(curlm, &num_transfers);\n+\t\tif (num_transfers < active_requests) {\n+\t\t\tprocess_curl_messages();\n+\t\t}\n+\t}\n+\n+\twhile (slot != NULL && slot->in_use) {\n+\t\tslot = slot->next;\n+\t}\n+\tif (slot == NULL) {\n+\t\tnewslot = xmalloc(sizeof(*newslot));\n+\t\tnewslot->curl = curl_easy_duphandle(curl_default);\n+\t\tnewslot->in_use = 0;\n+\t\tnewslot->next = NULL;\n+\n+\t\tslot = active_queue_head;\n+\t\tif (slot == NULL) {\n+\t\t\tactive_queue_head = newslot;\n+\t\t} else {\n+\t\t\twhile (slot->next != NULL) {\n+\t\t\t\tslot = slot->next;\n+\t\t\t}\n+\t\t\tslot->next = newslot;\n+\t\t}\n+\t\tslot = newslot;\n+\t}\n+\n+\tactive_requests++;\n+\tslot->in_use = 1;\n+\tslot->done = 0;\n+\tslot->local = NULL;\n+\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, no_pragma_header);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, no_range_header);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n+\n+\treturn slot;\n+}\n+\n+int start_active_slot(struct active_request_slot *slot)\n+{\n+\tCURLMcode curlm_result = curl_multi_add_handle(curlm, slot->curl);\n+\n+\tif (curlm_result != CURLM_OK &&\n+\t    curlm_result != CURLM_CALL_MULTI_PERFORM) {\n+\t\tactive_requests--;\n+\t\tslot->in_use = 0;\n+\t\treturn 0;\n+\t}\n+\n+\treturn 1;\n+}\n+\n+void run_active_slot(struct active_request_slot *slot)\n+{\n+\tint num_transfers;\n+\tlong last_pos = 0;\n+\tlong current_pos;\n+\tfd_set readfds;\n+\tfd_set writefds;\n+\tfd_set excfds;\n+\tint max_fd;\n+\tstruct timeval select_timeout;\n+\tCURLMcode curlm_result;\n+\n+\twhile (!slot->done) {\n+\t\tdata_received = 0;\n+\t\tdo {\n+\t\t\tcurlm_result = curl_multi_perform(curlm,\n+\t\t\t\t\t\t\t  &num_transfers);\n+\t\t} while (curlm_result == CURLM_CALL_MULTI_PERFORM);\n+\t\tif (num_transfers < active_requests) {\n+\t\t\tprocess_curl_messages();\n+\t\t\tprocess_request_queue();\n+\t\t}\n+\n+\t\tif (!data_received && slot->local != NULL) {\n+\t\t\tcurrent_pos = ftell(slot->local);\n+\t\t\tif (current_pos > last_pos)\n+\t\t\t\tdata_received++;\n+\t\t\tlast_pos = current_pos;\n+\t\t}\n+\n+\t\tif (!slot->done && !data_received) {\n+\t\t\tmax_fd = 0;\n+\t\t\tFD_ZERO(&readfds);\n+\t\t\tFD_ZERO(&writefds);\n+\t\t\tFD_ZERO(&excfds);\n+\t\t\tselect_timeout.tv_sec = 0;\n+\t\t\tselect_timeout.tv_usec = 50000;\n+\t\t\tselect(max_fd, &readfds, &writefds,\n+\t\t\t       &excfds, &select_timeout);\n+\t\t}\n+\t}\n+}\n+\n+void start_request(struct transfer_request *request)\n+{\n+\tchar *hex = sha1_to_hex(request->sha1);\n+\tchar prevfile[PATH_MAX];\n+\tchar *url;\n+\tchar *posn;\n+\tint prevlocal;\n+\tunsigned char prev_buf[PREV_BUF_SIZE];\n+\tssize_t prev_read = 0;\n+\tlong prev_posn = 0;\n+\tchar range[RANGE_HEADER_SIZE];\n+\tstruct curl_slist *range_header = NULL;\n+\tstruct active_request_slot *slot;\n+\n+\tsnprintf(prevfile, sizeof(prevfile), \"%s.prev\", request->filename);\n+\tunlink(prevfile);\n+\trename(request->tmpfile, prevfile);\n+\tunlink(request->tmpfile);\n+\n+\trequest->local = open(request->tmpfile,\n+\t\t\t      O_WRONLY | O_CREAT | O_EXCL, 0666);\n+\tif (request->local < 0) {\n+\t\trequest->state = ABORTED;\n+\t\terror(\"Couldn't create temporary file %s for %s: %s\\n\",\n+\t\t      request->tmpfile, request->filename, strerror(errno));\n+\t\treturn;\n+\t}\n+\n+\tmemset(&request->stream, 0, sizeof(request->stream));\n+\n+\tinflateInit(&request->stream);\n+\n+\tSHA1_Init(&request->c);\n+\n+\turl = xmalloc(strlen(request->repo->base) + 50);\n+\trequest->url = xmalloc(strlen(request->repo->base) + 50);\n+\tstrcpy(url, request->repo->base);\n+\tposn = url + strlen(request->repo->base);\n+\tstrcpy(posn, \"objects/\");\n+\tposn += 8;\n+\tmemcpy(posn, hex, 2);\n+\tposn += 2;\n+\t*(posn++) = '/';\n+\tstrcpy(posn, hex + 2);\n+\tstrcpy(request->url, url);\n+\n+\t/* If a previous temp file is present, process what was already\n+\t   fetched. */\n+\tprevlocal = open(prevfile, O_RDONLY);\n+\tif (prevlocal != -1) {\n+\t\tdo {\n+\t\t\tprev_read = read(prevlocal, prev_buf, PREV_BUF_SIZE);\n+\t\t\tif (prev_read>0) {\n+\t\t\t\tif (fwrite_sha1_file(prev_buf,\n+\t\t\t\t\t\t     1,\n+\t\t\t\t\t\t     prev_read,\n+\t\t\t\t\t\t     request) == prev_read) {\n+\t\t\t\t\tprev_posn += prev_read;\n+\t\t\t\t} else {\n+\t\t\t\t\tprev_read = -1;\n+\t\t\t\t}\n+\t\t\t}\n+\t\t} while (prev_read > 0);\n+\t\tclose(prevlocal);\n+\t}\n+\tunlink(prevfile);\n+\n+\t/* Reset inflate/SHA1 if there was an error reading the previous temp\n+\t   file; also rewind to the beginning of the local file. */\n+\tif (prev_read == -1) {\n+\t\tmemset(&request->stream, 0, sizeof(request->stream));\n+\t\tinflateInit(&request->stream);\n+\t\tSHA1_Init(&request->c);\n+\t\tif (prev_posn>0) {\n+\t\t\tprev_posn = 0;\n+\t\t\tlseek(request->local, SEEK_SET, 0);\n+\t\t\tftruncate(request->local, 0);\n+\t\t}\n+\t}\n+\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, request);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite_sha1_file);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_ERRORBUFFER, request->errorstr);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\n+\t/* If we have successfully processed data from a previous fetch\n+\t   attempt, only fetch the data we don't already have. */\n+\tif (prev_posn>0) {\n+\t\tif (get_verbosely)\n+\t\t\tfprintf(stderr,\n+\t\t\t\t\"Resuming fetch of object %s at byte %ld\\n\",\n+\t\t\t\thex, prev_posn);\n+\t\tsprintf(range, \"Range: bytes=%ld-\", prev_posn);\n+\t\trange_header = curl_slist_append(range_header, range);\n+\t\tcurl_easy_setopt(slot->curl,\n+\t\t\t\t CURLOPT_HTTPHEADER, range_header);\n+\t}\n+\n+\t/* Try to add to multi handle, abort the request on error */\n+\tif (!start_active_slot(slot)) {\n+\t\trequest->state = ABORTED;\n+\t\tclose(request->local);\n+\t\tfree(request->url);\n+\t\treturn;\n+\t}\n+\t\n+\trequest->slot = slot;\n+\trequest->state = ACTIVE;\n+}\n+\n+void finish_request(struct transfer_request *request)\n+{\n+\tfchmod(request->local, 0444);\n+\tclose(request->local);\n+\n+\tif (request->http_code == 416) {\n+\t\tfprintf(stderr, \"Warning: requested range invalid; we may already have all the data.\\n\");\n+\t} else if (request->curl_result != CURLE_OK) {\n+\t\treturn;\n+\t}\n+\n+\tinflateEnd(&request->stream);\n+\tSHA1_Final(request->real_sha1, &request->c);\n+\tif (request->zret != Z_STREAM_END) {\n+\t\tunlink(request->tmpfile);\n+\t\treturn;\n+\t}\n+\tif (memcmp(request->sha1, request->real_sha1, 20)) {\n+\t\tunlink(request->tmpfile);\n+\t\treturn;\n+\t}\n+\trequest->rename =\n+\t\trelink_or_rename(request->tmpfile, request->filename);\n+\n+\tif (request->rename == 0)\n+\t\tpull_say(\"got %s\\n\", sha1_to_hex(request->sha1));\n+}\n+\n+void release_request(struct transfer_request *request)\n+{\n+\tstruct transfer_request *entry = request_queue_head;\n+\n+\tif (request == request_queue_head) {\n+\t\trequest_queue_head = request->next;\n+\t} else {\n+\t\twhile (entry->next != NULL && entry->next != request)\n+\t\t\tentry = entry->next;\n+\t\tif (entry->next == request)\n+\t\t\tentry->next = entry->next->next;\n+\t}\n+\n+\tfree(request->url);\n+\tfree(request);\n+}\n+\n+void process_curl_messages()\n+{\n+\tint num_messages;\n+\tstruct active_request_slot *slot;\n+\tstruct transfer_request *request = NULL;\n+\tCURLMsg *curl_message = curl_multi_info_read(curlm, &num_messages);\n+\n+\twhile (curl_message != NULL) {\n+\t\tif (curl_message->msg == CURLMSG_DONE) {\n+\t\t\tslot = active_queue_head;\n+\t\t\twhile (slot != NULL &&\n+\t\t\t       slot->curl != curl_message->easy_handle)\n+\t\t\t\tslot = slot->next;\n+\t\t\tif (slot != NULL) {\n+\t\t\t\tcurl_multi_remove_handle(curlm, slot->curl);\n+\t\t\t\tactive_requests--;\n+\t\t\t\tslot->done = 1;\n+\t\t\t\tslot->in_use = 0;\n+\t\t\t\tslot->curl_result = curl_message->data.result;\n+\t\t\t\trequest = request_queue_head;\n+\t\t\t\twhile (request != NULL &&\n+\t\t\t\t       request->slot != slot)\n+\t\t\t\t\trequest = request->next;\n+\t\t\t} else {\n+\t\t\t\tfprintf(stderr, \"Received DONE message for unknown request!\\n\");\n+\t\t\t}\n+\t\t\tif (request != NULL) {\n+\t\t\t\trequest->curl_result =\n+\t\t\t\t\tcurl_message->data.result;\n+\t\t\t\tcurl_easy_getinfo(slot->curl,\n+\t\t\t\t\t\t  CURLINFO_HTTP_CODE,\n+\t\t\t\t\t\t  &request->http_code);\n+\t\t\t\trequest->slot = NULL;\n+\n+\t\t\t\t/* Use alternates if necessary */\n+\t\t\t\tif (request->http_code == 404 &&\n+\t\t\t\t    request->repo->next != NULL) {\n+\t\t\t\t\trequest->repo = request->repo->next;\n+\t\t\t\t\tstart_request(request);\n+\t\t\t\t} else {\n+\t\t\t\t\tfinish_request(request);\n+\t\t\t\t\trequest->state = COMPLETE;\n+\t\t\t\t}\n+\t\t\t}\n+\t\t} else {\n+\t\t\tfprintf(stderr, \"Unknown CURL message received: %d\\n\",\n+\t\t\t\t(int)curl_message->msg);\n+\t\t}\n+\t\tcurl_message = curl_multi_info_read(curlm, &num_messages);\n+\t}\n+}\n+\n+void process_request_queue()\n+{\n+\tstruct transfer_request *request = request_queue_head;\n+\tint num_transfers;\n+\n+\twhile (active_requests < max_requests && request != NULL) {\n+\t\tif (request->state == WAITING) {\n+\t\t\tstart_request(request);\n+\t\t\tcurl_multi_perform(curlm, &num_transfers);\n+\t\t}\n+\t\trequest = request->next;\n+\t}\n+}\n+\n+void prefetch(unsigned char *sha1)\n+{\n+\tstruct transfer_request *newreq;\n+\tstruct transfer_request *tail;\n+\tchar *filename = sha1_file_name(sha1);\n+\n+\tnewreq = xmalloc(sizeof(*newreq));\n+\tmemcpy(newreq->sha1, sha1, 20);\n+\tnewreq->repo = alt;\n+\tnewreq->url = NULL;\n+\tnewreq->local = -1;\n+\tnewreq->state = WAITING;\n+\tsnprintf(newreq->filename, sizeof(newreq->filename), \"%s\", filename);\n+\tsnprintf(newreq->tmpfile, sizeof(newreq->tmpfile),\n+\t\t \"%s.temp\", filename);\n+\tnewreq->next = NULL;\n+\n+\tif (request_queue_head == NULL) {\n+\t\trequest_queue_head = newreq;\n+\t} else {\n+\t\ttail = request_queue_head;\n+\t\twhile (tail->next != NULL) {\n+\t\t\ttail = tail->next;\n+\t\t}\n+\t\ttail->next = newreq;\n+\t}\n+\tprocess_request_queue();\n+\tprocess_curl_messages();\n+}\n+\n static int got_alternates = 0;\n \n static int fetch_index(struct alt_base *repo, unsigned char *sha1)\n {\n+\tchar *hex = sha1_to_hex(sha1);\n \tchar *filename;\n \tchar *url;\n \tchar tmpfile[PATH_MAX];\n@@ -121,20 +521,18 @@ static int fetch_index(struct alt_base *\n \tlong prev_posn = 0;\n \tchar range[RANGE_HEADER_SIZE];\n \tstruct curl_slist *range_header = NULL;\n-\tCURLcode curl_result;\n \n \tFILE *indexfile;\n+\tstruct active_request_slot *slot;\n \n \tif (has_pack_index(sha1))\n \t\treturn 0;\n \n \tif (get_verbosely)\n-\t\tfprintf(stderr, \"Getting index for pack %s\\n\",\n-\t\t\tsha1_to_hex(sha1));\n+\t\tfprintf(stderr, \"Getting index for pack %s\\n\", hex);\n \t\n \turl = xmalloc(strlen(repo->base) + 64);\n-\tsprintf(url, \"%s/objects/pack/pack-%s.idx\",\n-\t\trepo->base, sha1_to_hex(sha1));\n+\tsprintf(url, \"%s/objects/pack/pack-%s.idx\", repo->base, hex);\n \t\n \tfilename = sha1_pack_index_name(sha1);\n \tsnprintf(tmpfile, sizeof(tmpfile), \"%s.temp\", filename);\n@@ -143,12 +541,12 @@ static int fetch_index(struct alt_base *\n \t\treturn error(\"Unable to open local file %s for pack index\",\n \t\t\t     filename);\n \n-\tcurl_easy_setopt(curl, CURLOPT_FILE, indexfile);\n-\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite);\n-\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_pragma_header);\n-\tcurl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n-\t\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, indexfile);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\tslot->local = indexfile;\n+\n \t/* If there is data present from a previous transfer attempt,\n \t   resume where it left off */\n \tprev_posn = ftell(indexfile);\n@@ -156,20 +554,21 @@ static int fetch_index(struct alt_base *\n \t\tif (get_verbosely)\n \t\t\tfprintf(stderr,\n \t\t\t\t\"Resuming fetch of index for pack %s at byte %ld\\n\",\n-\t\t\t\tsha1_to_hex(sha1), prev_posn);\n+\t\t\t\thex, prev_posn);\n \t\tsprintf(range, \"Range: bytes=%ld-\", prev_posn);\n \t\trange_header = curl_slist_append(range_header, range);\n-\t\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, range_header);\n+\t\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, range_header);\n \t}\n \n-\t/* Clear out the Range: header after performing the request, so\n-\t   other curl requests don't inherit inappropriate header data */\n-\tcurl_result = curl_easy_perform(curl);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_range_header);\n-\tif (curl_result != 0) {\n-\t\tfclose(indexfile);\n-\t\treturn error(\"Unable to get pack index %s\\n%s\", url,\n-\t\t\t     curl_errorstr);\n+\tif (start_active_slot(slot)) {\n+\t\trun_active_slot(slot);\n+\t\tif (slot->curl_result != CURLE_OK) {\n+\t\t\tfclose(indexfile);\n+\t\t\treturn error(\"Unable to get pack index %s\\n%s\", url,\n+\t\t\t\t     curl_errorstr);\n+\t\t}\n+\t} else {\n+\t\treturn error(\"Unable to start request\");\n \t}\n \n \tfclose(indexfile);\n@@ -205,6 +604,9 @@ static int fetch_alternates(char *base)\n \tchar *data;\n \tint i = 0;\n \tint http_specific = 1;\n+\tstruct alt_base *tail = alt;\n+\n+\tstruct active_request_slot *slot;\n \tif (got_alternates)\n \t\treturn 0;\n \tdata = xmalloc(4096);\n@@ -218,22 +620,31 @@ static int fetch_alternates(char *base)\n \turl = xmalloc(strlen(base) + 31);\n \tsprintf(url, \"%s/objects/info/http-alternates\", base);\n \n-\tcurl_easy_setopt(curl, CURLOPT_FILE, &buffer);\n-\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n-\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n-\n-\tif (curl_easy_perform(curl) || !buffer.posn) {\n-\t\thttp_specific = 0;\n-\n-\t\tsprintf(url, \"%s/objects/info/alternates\", base);\n-\t\t\n-\t\tcurl_easy_setopt(curl, CURLOPT_FILE, &buffer);\n-\t\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n-\t\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n-\t\t\n-\t\tif (curl_easy_perform(curl)) {\n-\t\t\treturn 0;\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, &buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\tif (start_active_slot(slot)) {\n+\t\trun_active_slot(slot);\n+\t\tif (slot->curl_result != CURLE_OK || !buffer.posn) {\n+\t\t\thttp_specific = 0;\n+\n+\t\t\tsprintf(url, \"%s/objects/info/alternates\", base);\n+\n+\t\t\tslot = get_active_slot();\n+\t\t\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, &buffer);\n+\t\t\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION,\n+\t\t\t\t\t fwrite_buffer);\n+\t\t\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\t\t\tif (start_active_slot(slot)) {\n+\t\t\t\trun_active_slot(slot);\n+\t\t\t\tif (slot->curl_result != CURLE_OK) {\n+\t\t\t\t\treturn 0;\n+\t\t\t\t}\n+\t\t\t}\n \t\t}\n+\t} else {\n+\t\treturn 0;\n \t}\n \n \tdata[buffer.posn] = '\\0';\n@@ -283,11 +694,13 @@ static int fetch_alternates(char *base)\n \t\t\t\t\tfprintf(stderr, \n \t\t\t\t\t\t\"Also look at %s\\n\", target);\n \t\t\t\tnewalt = xmalloc(sizeof(*newalt));\n-\t\t\t\tnewalt->next = alt;\n+\t\t\t\tnewalt->next = NULL;\n \t\t\t\tnewalt->base = target;\n \t\t\t\tnewalt->got_indices = 0;\n \t\t\t\tnewalt->packs = NULL;\n-\t\t\t\talt = newalt;\n+\t\t\t\twhile (tail->next != NULL)\n+\t\t\t\t\ttail = tail->next;\n+\t\t\t\ttail->next = newalt;\n \t\t\t\tret++;\n \t\t\t}\n \t\t}\n@@ -306,6 +719,8 @@ static int fetch_indices(struct alt_base\n \tchar *data;\n \tint i = 0;\n \n+\tstruct active_request_slot *slot;\n+\n \tif (repo->got_indices)\n \t\treturn 0;\n \n@@ -320,14 +735,18 @@ static int fetch_indices(struct alt_base\n \turl = xmalloc(strlen(repo->base) + 21);\n \tsprintf(url, \"%s/objects/info/packs\", repo->base);\n \n-\tcurl_easy_setopt(curl, CURLOPT_FILE, &buffer);\n-\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n-\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, NULL);\n-\tcurl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n-\t\n-\tif (curl_easy_perform(curl))\n-\t\treturn error(\"%s\", curl_errorstr);\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, &buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, NULL);\n+\tif (start_active_slot(slot)) {\n+\t\trun_active_slot(slot);\n+\t\tif (slot->curl_result != CURLE_OK)\n+\t\t\treturn error(\"%s\", curl_errorstr);\n+\t} else {\n+\t\treturn error(\"Unable to start request\");\n+\t}\n \n \twhile (i < buffer.posn) {\n \t\tswitch (data[i]) {\n@@ -364,7 +783,8 @@ static int fetch_pack(struct alt_base *r\n \tlong prev_posn = 0;\n \tchar range[RANGE_HEADER_SIZE];\n \tstruct curl_slist *range_header = NULL;\n-\tCURLcode curl_result;\n+\n+\tstruct active_request_slot *slot;\n \n \tif (fetch_indices(repo))\n \t\treturn -1;\n@@ -390,11 +810,11 @@ static int fetch_pack(struct alt_base *r\n \t\treturn error(\"Unable to open local file %s for pack\",\n \t\t\t     filename);\n \n-\tcurl_easy_setopt(curl, CURLOPT_FILE, packfile);\n-\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite);\n-\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_pragma_header);\n-\tcurl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, packfile);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\tslot->local = packfile;\n \n \t/* If there is data present from a previous transfer attempt,\n \t   resume where it left off */\n@@ -406,17 +826,18 @@ static int fetch_pack(struct alt_base *r\n \t\t\t\tsha1_to_hex(target->sha1), prev_posn);\n \t\tsprintf(range, \"Range: bytes=%ld-\", prev_posn);\n \t\trange_header = curl_slist_append(range_header, range);\n-\t\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, range_header);\n+\t\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, range_header);\n \t}\n \n-\t/* Clear out the Range: header after performing the request, so\n-\t   other curl requests don't inherit inappropriate header data */\n-\tcurl_result = curl_easy_perform(curl);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_range_header);\n-\tif (curl_result != 0) {\n-\t\tfclose(packfile);\n-\t\treturn error(\"Unable to get pack file %s\\n%s\", url,\n-\t\t\t     curl_errorstr);\n+\tif (start_active_slot(slot)) {\n+\t\trun_active_slot(slot);\n+\t\tif (slot->curl_result != CURLE_OK) {\n+\t\t\tfclose(packfile);\n+\t\t\treturn error(\"Unable to get pack file %s\\n%s\", url,\n+\t\t\t\t     curl_errorstr);\n+\t\t}\n+\t} else {\n+\t\treturn error(\"Unable to start request\");\n \t}\n \n \tfclose(packfile);\n@@ -441,155 +862,73 @@ static int fetch_pack(struct alt_base *r\n static int fetch_object(struct alt_base *repo, unsigned char *sha1)\n {\n \tchar *hex = sha1_to_hex(sha1);\n-\tchar *filename = sha1_file_name(sha1);\n-\tunsigned char real_sha1[20];\n-\tchar tmpfile[PATH_MAX];\n-\tchar prevfile[PATH_MAX];\n \tint ret;\n-\tchar *url;\n-\tchar *posn;\n-\tint prevlocal;\n-\tunsigned char prev_buf[PREV_BUF_SIZE];\n-\tssize_t prev_read = 0;\n-\tlong prev_posn = 0;\n-\tchar range[RANGE_HEADER_SIZE];\n-\tstruct curl_slist *range_header = NULL;\n-\tCURLcode curl_result;\n-\n-\tsnprintf(tmpfile, sizeof(tmpfile), \"%s.temp\", filename);\n-\tsnprintf(prevfile, sizeof(prevfile), \"%s.prev\", filename);\n-\n-\tif (unlink(prevfile) && (errno != ENOENT))\n-\t\treturn error(\"Failed to unlink %s (%s)\",\n-\t\t\t     prevfile, strerror(errno));\n-\tif (rename(tmpfile, prevfile) && (errno != ENOENT))\n-\t\treturn error(\"Failed to rename %s to %s (%s)\",\n-\t\t\t     tmpfile, prevfile, strerror(errno));\n-\n-\tlocal = open(tmpfile, O_WRONLY | O_CREAT | O_EXCL, 0666);\n-\n-\t/* Note: if another instance starts now, it will turn our new\n-\t   tmpfile into its prevfile. */\n-\n-\tif (local < 0)\n-\t\treturn error(\"Couldn't create temporary file %s for %s: %s\\n\",\n-\t\t\t     tmpfile, filename, strerror(errno));\n+\tstruct transfer_request *request = request_queue_head;\n+\tint num_transfers;\n \n-\tmemset(&stream, 0, sizeof(stream));\n-\n-\tinflateInit(&stream);\n-\n-\tSHA1_Init(&c);\n-\n-\tcurl_easy_setopt(curl, CURLOPT_FAILONERROR, 1);\n-\tcurl_easy_setopt(curl, CURLOPT_FILE, NULL);\n-\tcurl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite_sha1_file);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_pragma_header);\n-\tcurl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n-\n-\turl = xmalloc(strlen(repo->base) + 50);\n-\tstrcpy(url, repo->base);\n-\tposn = url + strlen(repo->base);\n-\tstrcpy(posn, \"objects/\");\n-\tposn += 8;\n-\tmemcpy(posn, hex, 2);\n-\tposn += 2;\n-\t*(posn++) = '/';\n-\tstrcpy(posn, hex + 2);\n+\twhile (request != NULL && memcmp(request->sha1, sha1, 20))\n+\t\trequest = request->next;\n+\tif (request == NULL)\n+\t\treturn error(\"Couldn't find request for %s in the queue\", hex);\n+\n+\twhile (request->state == WAITING) {\n+\t\tcurl_multi_perform(curlm, &num_transfers);\n+\t\tif (num_transfers < active_requests) {\n+\t\t\tprocess_curl_messages();\n+\t\t\tprocess_request_queue();\n+\t\t}\n+\t}\n \n-\tcurl_easy_setopt(curl, CURLOPT_URL, url);\n+\tif (request->state == ACTIVE)\n+\t\trun_active_slot(request->slot);\n \n-\t/* If a previous temp file is present, process what was already\n-\t   fetched. */\n-\tprevlocal = open(prevfile, O_RDONLY);\n-\tif (prevlocal != -1) {\n-\t\tdo {\n-\t\t\tprev_read = read(prevlocal, prev_buf, PREV_BUF_SIZE);\n-\t\t\tif (prev_read>0) {\n-\t\t\t\tif (fwrite_sha1_file(prev_buf,\n-\t\t\t\t\t\t     1,\n-\t\t\t\t\t\t     prev_read,\n-\t\t\t\t\t\t     NULL) == prev_read) {\n-\t\t\t\t\tprev_posn += prev_read;\n-\t\t\t\t} else {\n-\t\t\t\t\tprev_read = -1;\n-\t\t\t\t}\n-\t\t\t}\n-\t\t} while (prev_read > 0);\n-\t\tclose(prevlocal);\n+\tif (request->state == ABORTED) {\n+\t\trelease_request(request);\n+\t\treturn error(\"Request for %s aborted\", hex);\n \t}\n-\tunlink(prevfile);\n \n-\t/* Reset inflate/SHA1 if there was an error reading the previous temp\n-\t   file; also rewind to the beginning of the local file. */\n-\tif (prev_read == -1) {\n-\t\tmemset(&stream, 0, sizeof(stream));\n-\t\tinflateInit(&stream);\n-\t\tSHA1_Init(&c);\n-\t\tif (prev_posn>0) {\n-\t\t\tprev_posn = 0;\n-\t\t\tlseek(local, SEEK_SET, 0);\n-\t\t\tftruncate(local, 0);\n-\t\t}\n+\tif (request->curl_result != CURLE_OK && request->http_code != 416) {\n+\t\tret = error(\"%s\", request->errorstr);\n+\t\trelease_request(request);\n+\t\treturn ret;\n \t}\n \n-\t/* If we have successfully processed data from a previous fetch\n-\t   attempt, only fetch the data we don't already have. */\n-\tif (prev_posn>0) {\n-\t\tif (get_verbosely)\n-\t\t\tfprintf(stderr,\n-\t\t\t\t\"Resuming fetch of object %s at byte %ld\\n\",\n-\t\t\t\thex, prev_posn);\n-\t\tsprintf(range, \"Range: bytes=%ld-\", prev_posn);\n-\t\trange_header = curl_slist_append(range_header, range);\n-\t\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, range_header);\n+\tif (request->zret != Z_STREAM_END) {\n+\t\tret = error(\"File %s (%s) corrupt\\n\", hex, request->url);\n+\t\trelease_request(request);\n+\t\treturn ret;\n \t}\n \n-\t/* Clear out the Range: header after performing the request, so\n-\t   other curl requests don't inherit inappropriate header data */\n-\tcurl_result = curl_easy_perform(curl);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, no_range_header);\n-\tif (curl_result != 0) {\n-\t\treturn error(\"%s\", curl_errorstr);\n-\t}\n-\n-\tfchmod(local, 0444);\n-\tclose(local);\n-\tinflateEnd(&stream);\n-\tSHA1_Final(real_sha1, &c);\n-\tif (zret != Z_STREAM_END) {\n-\t\tunlink(tmpfile);\n-\t\treturn error(\"File %s (%s) corrupt\\n\", hex, url);\n-\t}\n-\tif (memcmp(sha1, real_sha1, 20)) {\n-\t\tunlink(tmpfile);\n+\tif (memcmp(request->sha1, request->real_sha1, 20)) {\n+\t\trelease_request(request);\n \t\treturn error(\"File %s has bad hash\\n\", hex);\n \t}\n-\tret = relink_or_rename(tmpfile, filename);\n-\tif (ret)\n-\t\treturn error(\"unable to write sha1 filename %s: %s\",\n-\t\t\t     filename, strerror(ret));\n \n-\tpull_say(\"got %s\\n\", hex);\n+\tif (request->rename < 0) {\n+\t\tret = error(\"unable to write sha1 filename %s: %s\",\n+\t\t\t    request->filename,\n+\t\t\t    strerror(request->rename));\n+\t\trelease_request(request);\n+\t\treturn ret;\n+\t}\n+\n+\trelease_request(request);\n \treturn 0;\n }\n \n int fetch(unsigned char *sha1)\n {\n \tstruct alt_base *altbase = alt;\n+\n+\tif (!fetch_object(altbase, sha1))\n+\t\treturn 0;\n \twhile (altbase) {\n-\t\tif (!fetch_object(altbase, sha1))\n-\t\t\treturn 0;\n \t\tif (!fetch_pack(altbase, sha1))\n \t\t\treturn 0;\n-\t\tif (fetch_alternates(altbase->base) > 0) {\n-\t\t\taltbase = alt;\n-\t\t\tcontinue;\n-\t\t}\n \t\taltbase = altbase->next;\n \t}\n \treturn error(\"Unable to find %s under %s\\n\", sha1_to_hex(sha1), \n-\t\t     initial_base);\n+\t\t     alt->base);\n }\n \n int fetch_ref(char *ref, unsigned char *sha1)\n@@ -597,17 +936,13 @@ int fetch_ref(char *ref, unsigned char *\n         char *url, *posn;\n         char hex[42];\n         struct buffer buffer;\n-\tchar *base = initial_base;\n+\tchar *base = alt->base;\n+\tstruct active_request_slot *slot;\n         buffer.size = 41;\n         buffer.posn = 0;\n         buffer.buffer = hex;\n         hex[41] = '\\0';\n         \n-        curl_easy_setopt(curl, CURLOPT_FILE, &buffer);\n-        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n-\tcurl_easy_setopt(curl, CURLOPT_HTTPHEADER, NULL);\n-\tcurl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_errorstr);\n-\n         url = xmalloc(strlen(base) + 6 + strlen(ref));\n         strcpy(url, base);\n         posn = url + strlen(base);\n@@ -615,11 +950,19 @@ int fetch_ref(char *ref, unsigned char *\n         posn += 5;\n         strcpy(posn, ref);\n \n-        curl_easy_setopt(curl, CURLOPT_URL, url);\n-\n-        if (curl_easy_perform(curl))\n-                return error(\"Couldn't get %s for %s\\n%s\",\n-\t\t\t     url, ref, curl_errorstr);\n+\tslot = get_active_slot();\n+\tcurl_easy_setopt(slot->curl, CURLOPT_FILE, &buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_WRITEFUNCTION, fwrite_buffer);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_HTTPHEADER, NULL);\n+\tcurl_easy_setopt(slot->curl, CURLOPT_URL, url);\n+\tif (start_active_slot(slot)) {\n+\t\trun_active_slot(slot);\n+\t\tif (slot->curl_result != CURLE_OK)\n+\t\t\treturn error(\"Couldn't get %s for %s\\n%s\",\n+\t\t\t\t     url, ref, curl_errorstr);\n+\t} else {\n+\t\treturn error(\"Unable to start request\");\n+\t}\n \n         hex[40] = '\\0';\n         get_sha1_hex(hex, sha1);\n@@ -631,6 +974,7 @@ int main(int argc, char **argv)\n \tchar *commit_id;\n \tchar *url;\n \tint arg = 1;\n+\tstruct active_request_slot *slot;\n \n \twhile (arg < argc && argv[arg][0] == '-') {\n \t\tif (argv[arg][1] == 't') {\n@@ -648,6 +992,11 @@ int main(int argc, char **argv)\n \t\t\targ++;\n \t\t} else if (!strcmp(argv[arg], \"--recover\")) {\n \t\t\tget_recover = 1;\n+\t\t} else if (argv[arg][1] == 'r') {\n+\t\t\tmax_requests = atoi(argv[arg + 1]);\n+\t\t\tif (max_requests < 1)\n+\t\t\t\tmax_requests = DEFAULT_MAX_REQUESTS;\n+\t\t\targ++;\n \t\t}\n \t\targ++;\n \t}\n@@ -660,44 +1009,59 @@ int main(int argc, char **argv)\n \n \tcurl_global_init(CURL_GLOBAL_ALL);\n \n-\tcurl = curl_easy_init();\n+\tcurlm = curl_multi_init();\n+\tif (curlm == NULL) {\n+\t\tfprintf(stderr, \"Error creating curl multi handle.\\n\");\n+\t\treturn 1;\n+\t}\n \tno_pragma_header = curl_slist_append(no_pragma_header, \"Pragma:\");\n \tno_range_header = curl_slist_append(no_range_header, \"Range:\");\n \n+\tcurl_default = curl_easy_init();\n+\n \tcurl_ssl_verify = getenv(\"GIT_SSL_NO_VERIFY\") ? 0 : 1;\n-\tcurl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, curl_ssl_verify);\n+\tcurl_easy_setopt(curl_default, CURLOPT_SSL_VERIFYPEER, curl_ssl_verify);\n #if LIBCURL_VERSION_NUM >= 0x070907\n-\tcurl_easy_setopt(curl, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);\n+\tcurl_easy_setopt(curl_default, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);\n #endif\n \n \tif ((ssl_cert = getenv(\"GIT_SSL_CERT\")) != NULL) {\n-\t\tcurl_easy_setopt(curl, CURLOPT_SSLCERT, ssl_cert);\n+\t\tcurl_easy_setopt(curl_default, CURLOPT_SSLCERT, ssl_cert);\n \t}\n #if LIBCURL_VERSION_NUM >= 0x070902\n \tif ((ssl_key = getenv(\"GIT_SSL_KEY\")) != NULL) {\n-\t\tcurl_easy_setopt(curl, CURLOPT_SSLKEY, ssl_key);\n+\t\tcurl_easy_setopt(curl_default, CURLOPT_SSLKEY, ssl_key);\n \t}\n #endif\n #if LIBCURL_VERSION_NUM >= 0x070908\n \tif ((ssl_capath = getenv(\"GIT_SSL_CAPATH\")) != NULL) {\n-\t\tcurl_easy_setopt(curl, CURLOPT_CAPATH, ssl_capath);\n+\t\tcurl_easy_setopt(curl_default, CURLOPT_CAPATH, ssl_capath);\n \t}\n #endif\n \tif ((ssl_cainfo = getenv(\"GIT_SSL_CAINFO\")) != NULL) {\n-\t\tcurl_easy_setopt(curl, CURLOPT_CAINFO, ssl_cainfo);\n+\t\tcurl_easy_setopt(curl_default, CURLOPT_CAINFO, ssl_cainfo);\n \t}\n+\tcurl_easy_setopt(curl_default, CURLOPT_FAILONERROR, 1);\n \n \talt = xmalloc(sizeof(*alt));\n \talt->base = url;\n \talt->got_indices = 0;\n \talt->packs = NULL;\n \talt->next = NULL;\n-\tinitial_base = url;\n+\tfetch_alternates(alt->base);\n \n \tif (pull(commit_id))\n \t\treturn 1;\n \n \tcurl_slist_free_all(no_pragma_header);\n+\tcurl_slist_free_all(no_range_header);\n+\tcurl_easy_cleanup(curl_default);\n+\tslot = active_queue_head;\n+\twhile (slot != NULL) {\n+\t\tcurl_easy_cleanup(slot->curl);\n+\t\tslot = slot->next;\n+\t}\n+\tcurl_multi_cleanup(curlm);\n \tcurl_global_cleanup();\n \treturn 0;\n }\n"}]}