Merge branch 'jt/http-redact-cookies' into maint
[git.git] / streaming.c
blob5892b50bd89c3c66bdb541ca0100f0671834a542
1 /*
2 * Copyright (c) 2011, Google Inc.
3 */
4 #include "cache.h"
5 #include "streaming.h"
6 #include "packfile.h"
8 enum input_source {
9 stream_error = -1,
10 incore = 0,
11 loose = 1,
12 pack_non_delta = 2
15 typedef int (*open_istream_fn)(struct git_istream *,
16 struct object_info *,
17 const unsigned char *,
18 enum object_type *);
19 typedef int (*close_istream_fn)(struct git_istream *);
20 typedef ssize_t (*read_istream_fn)(struct git_istream *, char *, size_t);
22 struct stream_vtbl {
23 close_istream_fn close;
24 read_istream_fn read;
27 #define open_method_decl(name) \
28 int open_istream_ ##name \
29 (struct git_istream *st, struct object_info *oi, \
30 const unsigned char *sha1, \
31 enum object_type *type)
33 #define close_method_decl(name) \
34 int close_istream_ ##name \
35 (struct git_istream *st)
37 #define read_method_decl(name) \
38 ssize_t read_istream_ ##name \
39 (struct git_istream *st, char *buf, size_t sz)
41 /* forward declaration */
42 static open_method_decl(incore);
43 static open_method_decl(loose);
44 static open_method_decl(pack_non_delta);
45 static struct git_istream *attach_stream_filter(struct git_istream *st,
46 struct stream_filter *filter);
49 static open_istream_fn open_istream_tbl[] = {
50 open_istream_incore,
51 open_istream_loose,
52 open_istream_pack_non_delta,
55 #define FILTER_BUFFER (1024*16)
57 struct filtered_istream {
58 struct git_istream *upstream;
59 struct stream_filter *filter;
60 char ibuf[FILTER_BUFFER];
61 char obuf[FILTER_BUFFER];
62 int i_end, i_ptr;
63 int o_end, o_ptr;
64 int input_finished;
67 struct git_istream {
68 const struct stream_vtbl *vtbl;
69 unsigned long size; /* inflated size of full object */
70 git_zstream z;
71 enum { z_unused, z_used, z_done, z_error } z_state;
73 union {
74 struct {
75 char *buf; /* from read_object() */
76 unsigned long read_ptr;
77 } incore;
79 struct {
80 void *mapped;
81 unsigned long mapsize;
82 char hdr[32];
83 int hdr_avail;
84 int hdr_used;
85 } loose;
87 struct {
88 struct packed_git *pack;
89 off_t pos;
90 } in_pack;
92 struct filtered_istream filtered;
93 } u;
96 int close_istream(struct git_istream *st)
98 int r = st->vtbl->close(st);
99 free(st);
100 return r;
103 ssize_t read_istream(struct git_istream *st, void *buf, size_t sz)
105 return st->vtbl->read(st, buf, sz);
108 static enum input_source istream_source(const unsigned char *sha1,
109 enum object_type *type,
110 struct object_info *oi)
112 unsigned long size;
113 int status;
115 oi->typep = type;
116 oi->sizep = &size;
117 status = sha1_object_info_extended(sha1, oi, 0);
118 if (status < 0)
119 return stream_error;
121 switch (oi->whence) {
122 case OI_LOOSE:
123 return loose;
124 case OI_PACKED:
125 if (!oi->u.packed.is_delta && big_file_threshold < size)
126 return pack_non_delta;
127 /* fallthru */
128 default:
129 return incore;
133 struct git_istream *open_istream(const unsigned char *sha1,
134 enum object_type *type,
135 unsigned long *size,
136 struct stream_filter *filter)
138 struct git_istream *st;
139 struct object_info oi = OBJECT_INFO_INIT;
140 const unsigned char *real = lookup_replace_object(sha1);
141 enum input_source src = istream_source(real, type, &oi);
143 if (src < 0)
144 return NULL;
146 st = xmalloc(sizeof(*st));
147 if (open_istream_tbl[src](st, &oi, real, type)) {
148 if (open_istream_incore(st, &oi, real, type)) {
149 free(st);
150 return NULL;
153 if (filter) {
154 /* Add "&& !is_null_stream_filter(filter)" for performance */
155 struct git_istream *nst = attach_stream_filter(st, filter);
156 if (!nst) {
157 close_istream(st);
158 return NULL;
160 st = nst;
163 *size = st->size;
164 return st;
168 /*****************************************************************
170 * Common helpers
172 *****************************************************************/
174 static void close_deflated_stream(struct git_istream *st)
176 if (st->z_state == z_used)
177 git_inflate_end(&st->z);
181 /*****************************************************************
183 * Filtered stream
185 *****************************************************************/
187 static close_method_decl(filtered)
189 free_stream_filter(st->u.filtered.filter);
190 return close_istream(st->u.filtered.upstream);
193 static read_method_decl(filtered)
195 struct filtered_istream *fs = &(st->u.filtered);
196 size_t filled = 0;
198 while (sz) {
199 /* do we already have filtered output? */
200 if (fs->o_ptr < fs->o_end) {
201 size_t to_move = fs->o_end - fs->o_ptr;
202 if (sz < to_move)
203 to_move = sz;
204 memcpy(buf + filled, fs->obuf + fs->o_ptr, to_move);
205 fs->o_ptr += to_move;
206 sz -= to_move;
207 filled += to_move;
208 continue;
210 fs->o_end = fs->o_ptr = 0;
212 /* do we have anything to feed the filter with? */
213 if (fs->i_ptr < fs->i_end) {
214 size_t to_feed = fs->i_end - fs->i_ptr;
215 size_t to_receive = FILTER_BUFFER;
216 if (stream_filter(fs->filter,
217 fs->ibuf + fs->i_ptr, &to_feed,
218 fs->obuf, &to_receive))
219 return -1;
220 fs->i_ptr = fs->i_end - to_feed;
221 fs->o_end = FILTER_BUFFER - to_receive;
222 continue;
225 /* tell the filter to drain upon no more input */
226 if (fs->input_finished) {
227 size_t to_receive = FILTER_BUFFER;
228 if (stream_filter(fs->filter,
229 NULL, NULL,
230 fs->obuf, &to_receive))
231 return -1;
232 fs->o_end = FILTER_BUFFER - to_receive;
233 if (!fs->o_end)
234 break;
235 continue;
237 fs->i_end = fs->i_ptr = 0;
239 /* refill the input from the upstream */
240 if (!fs->input_finished) {
241 fs->i_end = read_istream(fs->upstream, fs->ibuf, FILTER_BUFFER);
242 if (fs->i_end < 0)
243 return -1;
244 if (fs->i_end)
245 continue;
247 fs->input_finished = 1;
249 return filled;
252 static struct stream_vtbl filtered_vtbl = {
253 close_istream_filtered,
254 read_istream_filtered,
257 static struct git_istream *attach_stream_filter(struct git_istream *st,
258 struct stream_filter *filter)
260 struct git_istream *ifs = xmalloc(sizeof(*ifs));
261 struct filtered_istream *fs = &(ifs->u.filtered);
263 ifs->vtbl = &filtered_vtbl;
264 fs->upstream = st;
265 fs->filter = filter;
266 fs->i_end = fs->i_ptr = 0;
267 fs->o_end = fs->o_ptr = 0;
268 fs->input_finished = 0;
269 ifs->size = -1; /* unknown */
270 return ifs;
273 /*****************************************************************
275 * Loose object stream
277 *****************************************************************/
279 static read_method_decl(loose)
281 size_t total_read = 0;
283 switch (st->z_state) {
284 case z_done:
285 return 0;
286 case z_error:
287 return -1;
288 default:
289 break;
292 if (st->u.loose.hdr_used < st->u.loose.hdr_avail) {
293 size_t to_copy = st->u.loose.hdr_avail - st->u.loose.hdr_used;
294 if (sz < to_copy)
295 to_copy = sz;
296 memcpy(buf, st->u.loose.hdr + st->u.loose.hdr_used, to_copy);
297 st->u.loose.hdr_used += to_copy;
298 total_read += to_copy;
301 while (total_read < sz) {
302 int status;
304 st->z.next_out = (unsigned char *)buf + total_read;
305 st->z.avail_out = sz - total_read;
306 status = git_inflate(&st->z, Z_FINISH);
308 total_read = st->z.next_out - (unsigned char *)buf;
310 if (status == Z_STREAM_END) {
311 git_inflate_end(&st->z);
312 st->z_state = z_done;
313 break;
315 if (status != Z_OK && (status != Z_BUF_ERROR || total_read < sz)) {
316 git_inflate_end(&st->z);
317 st->z_state = z_error;
318 return -1;
321 return total_read;
324 static close_method_decl(loose)
326 close_deflated_stream(st);
327 munmap(st->u.loose.mapped, st->u.loose.mapsize);
328 return 0;
331 static struct stream_vtbl loose_vtbl = {
332 close_istream_loose,
333 read_istream_loose,
336 static open_method_decl(loose)
338 st->u.loose.mapped = map_sha1_file(sha1, &st->u.loose.mapsize);
339 if (!st->u.loose.mapped)
340 return -1;
341 if ((unpack_sha1_header(&st->z,
342 st->u.loose.mapped,
343 st->u.loose.mapsize,
344 st->u.loose.hdr,
345 sizeof(st->u.loose.hdr)) < 0) ||
346 (parse_sha1_header(st->u.loose.hdr, &st->size) < 0)) {
347 git_inflate_end(&st->z);
348 munmap(st->u.loose.mapped, st->u.loose.mapsize);
349 return -1;
352 st->u.loose.hdr_used = strlen(st->u.loose.hdr) + 1;
353 st->u.loose.hdr_avail = st->z.total_out;
354 st->z_state = z_used;
356 st->vtbl = &loose_vtbl;
357 return 0;
361 /*****************************************************************
363 * Non-delta packed object stream
365 *****************************************************************/
367 static read_method_decl(pack_non_delta)
369 size_t total_read = 0;
371 switch (st->z_state) {
372 case z_unused:
373 memset(&st->z, 0, sizeof(st->z));
374 git_inflate_init(&st->z);
375 st->z_state = z_used;
376 break;
377 case z_done:
378 return 0;
379 case z_error:
380 return -1;
381 case z_used:
382 break;
385 while (total_read < sz) {
386 int status;
387 struct pack_window *window = NULL;
388 unsigned char *mapped;
390 mapped = use_pack(st->u.in_pack.pack, &window,
391 st->u.in_pack.pos, &st->z.avail_in);
393 st->z.next_out = (unsigned char *)buf + total_read;
394 st->z.avail_out = sz - total_read;
395 st->z.next_in = mapped;
396 status = git_inflate(&st->z, Z_FINISH);
398 st->u.in_pack.pos += st->z.next_in - mapped;
399 total_read = st->z.next_out - (unsigned char *)buf;
400 unuse_pack(&window);
402 if (status == Z_STREAM_END) {
403 git_inflate_end(&st->z);
404 st->z_state = z_done;
405 break;
407 if (status != Z_OK && status != Z_BUF_ERROR) {
408 git_inflate_end(&st->z);
409 st->z_state = z_error;
410 return -1;
413 return total_read;
416 static close_method_decl(pack_non_delta)
418 close_deflated_stream(st);
419 return 0;
422 static struct stream_vtbl pack_non_delta_vtbl = {
423 close_istream_pack_non_delta,
424 read_istream_pack_non_delta,
427 static open_method_decl(pack_non_delta)
429 struct pack_window *window;
430 enum object_type in_pack_type;
432 st->u.in_pack.pack = oi->u.packed.pack;
433 st->u.in_pack.pos = oi->u.packed.offset;
434 window = NULL;
436 in_pack_type = unpack_object_header(st->u.in_pack.pack,
437 &window,
438 &st->u.in_pack.pos,
439 &st->size);
440 unuse_pack(&window);
441 switch (in_pack_type) {
442 default:
443 return -1; /* we do not do deltas for now */
444 case OBJ_COMMIT:
445 case OBJ_TREE:
446 case OBJ_BLOB:
447 case OBJ_TAG:
448 break;
450 st->z_state = z_unused;
451 st->vtbl = &pack_non_delta_vtbl;
452 return 0;
456 /*****************************************************************
458 * In-core stream
460 *****************************************************************/
462 static close_method_decl(incore)
464 free(st->u.incore.buf);
465 return 0;
468 static read_method_decl(incore)
470 size_t read_size = sz;
471 size_t remainder = st->size - st->u.incore.read_ptr;
473 if (remainder <= read_size)
474 read_size = remainder;
475 if (read_size) {
476 memcpy(buf, st->u.incore.buf + st->u.incore.read_ptr, read_size);
477 st->u.incore.read_ptr += read_size;
479 return read_size;
482 static struct stream_vtbl incore_vtbl = {
483 close_istream_incore,
484 read_istream_incore,
487 static open_method_decl(incore)
489 st->u.incore.buf = read_sha1_file_extended(sha1, type, &st->size, 0);
490 st->u.incore.read_ptr = 0;
491 st->vtbl = &incore_vtbl;
493 return st->u.incore.buf ? 0 : -1;
497 /****************************************************************
498 * Users of streaming interface
499 ****************************************************************/
501 int stream_blob_to_fd(int fd, const struct object_id *oid, struct stream_filter *filter,
502 int can_seek)
504 struct git_istream *st;
505 enum object_type type;
506 unsigned long sz;
507 ssize_t kept = 0;
508 int result = -1;
510 st = open_istream(oid->hash, &type, &sz, filter);
511 if (!st) {
512 if (filter)
513 free_stream_filter(filter);
514 return result;
516 if (type != OBJ_BLOB)
517 goto close_and_exit;
518 for (;;) {
519 char buf[1024 * 16];
520 ssize_t wrote, holeto;
521 ssize_t readlen = read_istream(st, buf, sizeof(buf));
523 if (readlen < 0)
524 goto close_and_exit;
525 if (!readlen)
526 break;
527 if (can_seek && sizeof(buf) == readlen) {
528 for (holeto = 0; holeto < readlen; holeto++)
529 if (buf[holeto])
530 break;
531 if (readlen == holeto) {
532 kept += holeto;
533 continue;
537 if (kept && lseek(fd, kept, SEEK_CUR) == (off_t) -1)
538 goto close_and_exit;
539 else
540 kept = 0;
541 wrote = write_in_full(fd, buf, readlen);
543 if (wrote < 0)
544 goto close_and_exit;
546 if (kept && (lseek(fd, kept - 1, SEEK_CUR) == (off_t) -1 ||
547 xwrite(fd, "", 1) != 1))
548 goto close_and_exit;
549 result = 0;
551 close_and_exit:
552 close_istream(st);
553 return result;