Commit 7a520f5d by Carlos Martín Nieto

fetch: use the streaming indexer when downloading a pack

This changes the git_remote_download() API, but the existing one is
silly, so you don't get to complain.

The new API allows to know how much data has been downloaded, how many
objects we expect in total and how many we've processed.
parent f9f2344b
...@@ -150,7 +150,7 @@ GIT_EXTERN(int) git_remote_ls(git_remote *remote, git_headlist_cb list_cb, void ...@@ -150,7 +150,7 @@ GIT_EXTERN(int) git_remote_ls(git_remote *remote, git_headlist_cb list_cb, void
* @param filename where to store the temproray filename * @param filename where to store the temproray filename
* @return GIT_SUCCESS or an error code * @return GIT_SUCCESS or an error code
*/ */
GIT_EXTERN(int) git_remote_download(char **filename, git_remote *remote); GIT_EXTERN(int) git_remote_download(git_remote *remote, git_off_t *bytes, git_indexer_stats *stats);
/** /**
* Check whether the remote is connected * Check whether the remote is connected
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "git2/oid.h" #include "git2/oid.h"
#include "git2/refs.h" #include "git2/refs.h"
#include "git2/revwalk.h" #include "git2/revwalk.h"
#include "git2/indexer.h"
#include "common.h" #include "common.h"
#include "transport.h" #include "transport.h"
...@@ -101,30 +102,27 @@ int git_fetch_negotiate(git_remote *remote) ...@@ -101,30 +102,27 @@ int git_fetch_negotiate(git_remote *remote)
return t->negotiate_fetch(t, remote->repo, &remote->refs); return t->negotiate_fetch(t, remote->repo, &remote->refs);
} }
int git_fetch_download_pack(char **out, git_remote *remote) int git_fetch_download_pack(git_remote *remote, git_off_t *bytes, git_indexer_stats *stats)
{ {
if(!remote->need_pack) { if(!remote->need_pack)
*out = NULL;
return 0; return 0;
}
return remote->transport->download_pack(out, remote->transport, remote->repo); return remote->transport->download_pack(remote->transport, remote->repo, bytes, stats);
} }
/* Receiving data from a socket and storing it is pretty much the same for git and HTTP */ /* Receiving data from a socket and storing it is pretty much the same for git and HTTP */
int git_fetch__download_pack( int git_fetch__download_pack(
char **out,
const char *buffered, const char *buffered,
size_t buffered_size, size_t buffered_size,
GIT_SOCKET fd, GIT_SOCKET fd,
git_repository *repo) git_repository *repo,
git_off_t *bytes,
git_indexer_stats *stats)
{ {
git_filebuf file = GIT_FILEBUF_INIT; int recvd;
int error;
char buff[1024]; char buff[1024];
git_buf path = GIT_BUF_INIT;
static const char suff[] = "/objects/pack/pack-received";
gitno_buffer buf; gitno_buffer buf;
git_indexer_stream *idx;
gitno_buffer_setup(&buf, buff, sizeof(buff), fd); gitno_buffer_setup(&buf, buff, sizeof(buff), fd);
...@@ -133,41 +131,33 @@ int git_fetch__download_pack( ...@@ -133,41 +131,33 @@ int git_fetch__download_pack(
return -1; return -1;
} }
if (git_buf_joinpath(&path, repo->path_repository, suff) < 0) if (git_indexer_stream_new(&idx, git_repository_path(repo)) < 0)
goto on_error; return -1;
if (git_filebuf_open(&file, path.ptr, GIT_FILEBUF_TEMPORARY) < 0) memset(stats, 0, sizeof(git_indexer_stats));
if (git_indexer_stream_add(idx, buffered, buffered_size, stats) < 0)
goto on_error; goto on_error;
/* Part of the packfile has been received, don't loose it */ *bytes = buffered_size;
if (git_filebuf_write(&file, buffered, buffered_size) < 0)
goto on_error;
while (1) { do {
if (git_filebuf_write(&file, buf.data, buf.offset) < 0) if (git_indexer_stream_add(idx, buf.data, buf.offset, stats) < 0)
goto on_error; goto on_error;
gitno_consume_n(&buf, buf.offset); gitno_consume_n(&buf, buf.offset);
error = gitno_recv(&buf); if ((recvd = gitno_recv(&buf)) < 0)
if (error < GIT_SUCCESS)
goto on_error; goto on_error;
if (error == 0) /* Orderly shutdown */
break;
}
*out = git__strdup(file.path_lock); *bytes += recvd;
if (*out == NULL) } while(recvd > 0);
goto on_error;
/* A bit dodgy, but we need to keep the pack at the temporary path */ if (git_indexer_stream_finalize(idx, stats))
if (git_filebuf_commit_at(&file, file.path_lock, GIT_PACK_FILE_MODE) < 0)
goto on_error; goto on_error;
git_buf_free(&path); git_indexer_stream_free(idx);
return 0; return 0;
on_error: on_error:
git_buf_free(&path); git_indexer_stream_free(idx);
git_filebuf_cleanup(&file);
return -1; return -1;
} }
...@@ -10,9 +10,9 @@ ...@@ -10,9 +10,9 @@
#include "netops.h" #include "netops.h"
int git_fetch_negotiate(git_remote *remote); int git_fetch_negotiate(git_remote *remote);
int git_fetch_download_pack(char **out, git_remote *remote); int git_fetch_download_pack(git_remote *remote, git_off_t *bytes, git_indexer_stats *stats);
int git_fetch__download_pack(char **out, const char *buffered, size_t buffered_size, int git_fetch__download_pack(const char *buffered, size_t buffered_size, GIT_SOCKET fd,
GIT_SOCKET fd, git_repository *repo); git_repository *repo, git_off_t *bytes, git_indexer_stats *stats);
#endif #endif
...@@ -297,16 +297,16 @@ int git_remote_ls(git_remote *remote, git_headlist_cb list_cb, void *payload) ...@@ -297,16 +297,16 @@ int git_remote_ls(git_remote *remote, git_headlist_cb list_cb, void *payload)
return remote->transport->ls(remote->transport, list_cb, payload); return remote->transport->ls(remote->transport, list_cb, payload);
} }
int git_remote_download(char **filename, git_remote *remote) int git_remote_download(git_remote *remote, git_off_t *bytes, git_indexer_stats *stats)
{ {
int error; int error;
assert(filename && remote); assert(remote && bytes && stats);
if ((error = git_fetch_negotiate(remote)) < 0) if ((error = git_fetch_negotiate(remote)) < 0)
return error; return error;
return git_fetch_download_pack(filename, remote); return git_fetch_download_pack(remote, bytes, stats);
} }
int git_remote_update_tips(git_remote *remote) int git_remote_update_tips(git_remote *remote)
......
...@@ -81,7 +81,7 @@ struct git_transport { ...@@ -81,7 +81,7 @@ struct git_transport {
/** /**
* Download the packfile * Download the packfile
*/ */
int (*download_pack)(char **out, struct git_transport *transport, git_repository *repo); int (*download_pack)(struct git_transport *transport, git_repository *repo, git_off_t *bytes, git_indexer_stats *stats);
/** /**
* Fetch the changes * Fetch the changes
*/ */
......
...@@ -382,7 +382,7 @@ static int git_send_done(git_transport *transport) ...@@ -382,7 +382,7 @@ static int git_send_done(git_transport *transport)
return git_pkt_send_done(t->socket); return git_pkt_send_done(t->socket);
} }
static int git_download_pack(char **out, git_transport *transport, git_repository *repo) static int git_download_pack(git_transport *transport, git_repository *repo, git_off_t *bytes, git_indexer_stats *stats)
{ {
transport_git *t = (transport_git *) transport; transport_git *t = (transport_git *) transport;
int error = 0, read_bytes; int error = 0, read_bytes;
...@@ -410,7 +410,7 @@ static int git_download_pack(char **out, git_transport *transport, git_repositor ...@@ -410,7 +410,7 @@ static int git_download_pack(char **out, git_transport *transport, git_repositor
if (pkt->type == GIT_PKT_PACK) { if (pkt->type == GIT_PKT_PACK) {
git__free(pkt); git__free(pkt);
return git_fetch__download_pack(out, buf->data, buf->offset, t->socket, repo); return git_fetch__download_pack(buf->data, buf->offset, t->socket, repo, bytes, stats);
} }
/* For now we don't care about anything */ /* For now we don't care about anything */
......
...@@ -529,7 +529,8 @@ cleanup: ...@@ -529,7 +529,8 @@ cleanup:
} }
typedef struct { typedef struct {
git_filebuf *file; git_indexer_stream *idx;
git_indexer_stats *stats;
transport_http *transport; transport_http *transport;
} download_pack_cbdata; } download_pack_cbdata;
...@@ -545,10 +546,10 @@ static int on_body_download_pack(http_parser *parser, const char *str, size_t le ...@@ -545,10 +546,10 @@ static int on_body_download_pack(http_parser *parser, const char *str, size_t le
{ {
download_pack_cbdata *data = (download_pack_cbdata *) parser->data; download_pack_cbdata *data = (download_pack_cbdata *) parser->data;
transport_http *t = data->transport; transport_http *t = data->transport;
git_filebuf *file = data->file; git_indexer_stream *idx = data->idx;
git_indexer_stats *stats = data->stats;
return t->error = git_indexer_stream_add(idx, str, len, stats);
return t->error = git_filebuf_write(file, str, len);
} }
/* /*
...@@ -557,80 +558,68 @@ static int on_body_download_pack(http_parser *parser, const char *str, size_t le ...@@ -557,80 +558,68 @@ static int on_body_download_pack(http_parser *parser, const char *str, size_t le
* the simple downloader. Furthermore, we're using keep-alive * the simple downloader. Furthermore, we're using keep-alive
* connections, so the simple downloader would just hang. * connections, so the simple downloader would just hang.
*/ */
static int http_download_pack(char **out, git_transport *transport, git_repository *repo) static int http_download_pack(git_transport *transport, git_repository *repo, git_off_t *bytes, git_indexer_stats *stats)
{ {
transport_http *t = (transport_http *) transport; transport_http *t = (transport_http *) transport;
git_buf *oldbuf = &t->buf; git_buf *oldbuf = &t->buf;
int ret = 0; int recvd;
http_parser_settings settings; http_parser_settings settings;
char buffer[1024]; char buffer[1024];
gitno_buffer buf; gitno_buffer buf;
git_indexer_stream *idx = NULL;
download_pack_cbdata data; download_pack_cbdata data;
git_filebuf file = GIT_FILEBUF_INIT;
git_buf path = GIT_BUF_INIT; gitno_buffer_setup(&buf, buffer, sizeof(buffer), t->socket);
char suff[] = "/objects/pack/pack-received\0";
if (memcmp(oldbuf->ptr, "PACK", strlen("PACK"))) {
giterr_set(GITERR_NET, "The pack doesn't start with a pack signature");
return -1;
}
if (git_indexer_stream_new(&idx, git_repository_path(repo)) < 0)
return -1;
/* /*
* This is part of the previous response, so we don't want to * This is part of the previous response, so we don't want to
* re-init the parser, just set these two callbacks. * re-init the parser, just set these two callbacks.
*/ */
data.file = &file; memset(stats, 0, sizeof(git_indexer_stats));
data.stats = stats;
data.idx = idx;
data.transport = t; data.transport = t;
t->parser.data = &data; t->parser.data = &data;
t->transfer_finished = 0; t->transfer_finished = 0;
memset(&settings, 0x0, sizeof(settings)); memset(&settings, 0x0, sizeof(settings));
settings.on_message_complete = on_message_complete_download_pack; settings.on_message_complete = on_message_complete_download_pack;
settings.on_body = on_body_download_pack; settings.on_body = on_body_download_pack;
*bytes = oldbuf->size;
gitno_buffer_setup(&buf, buffer, sizeof(buffer), t->socket); if (git_indexer_stream_add(idx, oldbuf->ptr, oldbuf->size, stats) < 0)
if (memcmp(oldbuf->ptr, "PACK", strlen("PACK"))) {
giterr_set(GITERR_NET, "The pack doesn't start with a pack signature");
return -1;
}
if (git_buf_joinpath(&path, repo->path_repository, suff) < 0)
goto on_error;
if (git_filebuf_open(&file, path.ptr, GIT_FILEBUF_TEMPORARY) < 0)
goto on_error;
/* Part of the packfile has been received, don't loose it */
if (git_filebuf_write(&file, oldbuf->ptr, oldbuf->size) < 0)
goto on_error; goto on_error;
while(1) { do {
size_t parsed; size_t parsed;
ret = gitno_recv(&buf); if ((recvd = gitno_recv(&buf)) < 0)
if (ret < 0)
goto on_error; goto on_error;
parsed = http_parser_execute(&t->parser, &settings, buf.data, buf.offset); parsed = http_parser_execute(&t->parser, &settings, buf.data, buf.offset);
/* Both should happen at the same time */
if (parsed != buf.offset || t->error < 0) if (parsed != buf.offset || t->error < 0)
return t->error; goto on_error;
*bytes += recvd;
gitno_consume_n(&buf, parsed); gitno_consume_n(&buf, parsed);
} while (recvd > 0 && !t->transfer_finished);
if (ret == 0 || t->transfer_finished) { if (git_indexer_stream_finalize(idx, stats) < 0)
break; goto on_error;
}
}
*out = git__strdup(file.path_lock);
GITERR_CHECK_ALLOC(*out);
/* A bit dodgy, but we need to keep the pack at the temporary path */
ret = git_filebuf_commit_at(&file, file.path_lock, GIT_PACK_FILE_MODE);
git_buf_free(&path);
git_indexer_stream_free(idx);
return 0; return 0;
on_error: on_error:
git_filebuf_cleanup(&file); git_indexer_stream_free(idx);
git_buf_free(&path);
return -1; return -1;
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment