userdiff style: declare patterns with consistent style
[git] / streaming.c
1 /*
2  * Copyright (c) 2011, Google Inc.
3  */
4 #include "cache.h"
5 #include "streaming.h"
6 #include "repository.h"
7 #include "object-store.h"
8 #include "replace-object.h"
9 #include "packfile.h"
10
11 enum input_source {
12         stream_error = -1,
13         incore = 0,
14         loose = 1,
15         pack_non_delta = 2
16 };
17
18 typedef int (*open_istream_fn)(struct git_istream *,
19                                struct repository *,
20                                struct object_info *,
21                                const struct object_id *,
22                                enum object_type *);
23 typedef int (*close_istream_fn)(struct git_istream *);
24 typedef ssize_t (*read_istream_fn)(struct git_istream *, char *, size_t);
25
26 struct stream_vtbl {
27         close_istream_fn close;
28         read_istream_fn read;
29 };
30
31 #define open_method_decl(name) \
32         int open_istream_ ##name \
33         (struct git_istream *st, struct repository *r, \
34          struct object_info *oi, const struct object_id *oid, \
35          enum object_type *type)
36
37 #define close_method_decl(name) \
38         int close_istream_ ##name \
39         (struct git_istream *st)
40
41 #define read_method_decl(name) \
42         ssize_t read_istream_ ##name \
43         (struct git_istream *st, char *buf, size_t sz)
44
45 /* forward declaration */
46 static open_method_decl(incore);
47 static open_method_decl(loose);
48 static open_method_decl(pack_non_delta);
49 static struct git_istream *attach_stream_filter(struct git_istream *st,
50                                                 struct stream_filter *filter);
51
52
53 static open_istream_fn open_istream_tbl[] = {
54         open_istream_incore,
55         open_istream_loose,
56         open_istream_pack_non_delta,
57 };
58
59 #define FILTER_BUFFER (1024*16)
60
61 struct filtered_istream {
62         struct git_istream *upstream;
63         struct stream_filter *filter;
64         char ibuf[FILTER_BUFFER];
65         char obuf[FILTER_BUFFER];
66         int i_end, i_ptr;
67         int o_end, o_ptr;
68         int input_finished;
69 };
70
71 struct git_istream {
72         const struct stream_vtbl *vtbl;
73         unsigned long size; /* inflated size of full object */
74         git_zstream z;
75         enum { z_unused, z_used, z_done, z_error } z_state;
76
77         union {
78                 struct {
79                         char *buf; /* from read_object() */
80                         unsigned long read_ptr;
81                 } incore;
82
83                 struct {
84                         void *mapped;
85                         unsigned long mapsize;
86                         char hdr[32];
87                         int hdr_avail;
88                         int hdr_used;
89                 } loose;
90
91                 struct {
92                         struct packed_git *pack;
93                         off_t pos;
94                 } in_pack;
95
96                 struct filtered_istream filtered;
97         } u;
98 };
99
100 int close_istream(struct git_istream *st)
101 {
102         int r = st->vtbl->close(st);
103         free(st);
104         return r;
105 }
106
107 ssize_t read_istream(struct git_istream *st, void *buf, size_t sz)
108 {
109         return st->vtbl->read(st, buf, sz);
110 }
111
112 static enum input_source istream_source(struct repository *r,
113                                         const struct object_id *oid,
114                                         enum object_type *type,
115                                         struct object_info *oi)
116 {
117         unsigned long size;
118         int status;
119
120         oi->typep = type;
121         oi->sizep = &size;
122         status = oid_object_info_extended(r, oid, oi, 0);
123         if (status < 0)
124                 return stream_error;
125
126         switch (oi->whence) {
127         case OI_LOOSE:
128                 return loose;
129         case OI_PACKED:
130                 if (!oi->u.packed.is_delta && big_file_threshold < size)
131                         return pack_non_delta;
132                 /* fallthru */
133         default:
134                 return incore;
135         }
136 }
137
138 struct git_istream *open_istream(struct repository *r,
139                                  const struct object_id *oid,
140                                  enum object_type *type,
141                                  unsigned long *size,
142                                  struct stream_filter *filter)
143 {
144         struct git_istream *st;
145         struct object_info oi = OBJECT_INFO_INIT;
146         const struct object_id *real = lookup_replace_object(r, oid);
147         enum input_source src = istream_source(r, real, type, &oi);
148
149         if (src < 0)
150                 return NULL;
151
152         st = xmalloc(sizeof(*st));
153         if (open_istream_tbl[src](st, r, &oi, real, type)) {
154                 if (open_istream_incore(st, r, &oi, real, type)) {
155                         free(st);
156                         return NULL;
157                 }
158         }
159         if (filter) {
160                 /* Add "&& !is_null_stream_filter(filter)" for performance */
161                 struct git_istream *nst = attach_stream_filter(st, filter);
162                 if (!nst) {
163                         close_istream(st);
164                         return NULL;
165                 }
166                 st = nst;
167         }
168
169         *size = st->size;
170         return st;
171 }
172
173
174 /*****************************************************************
175  *
176  * Common helpers
177  *
178  *****************************************************************/
179
180 static void close_deflated_stream(struct git_istream *st)
181 {
182         if (st->z_state == z_used)
183                 git_inflate_end(&st->z);
184 }
185
186
187 /*****************************************************************
188  *
189  * Filtered stream
190  *
191  *****************************************************************/
192
193 static close_method_decl(filtered)
194 {
195         free_stream_filter(st->u.filtered.filter);
196         return close_istream(st->u.filtered.upstream);
197 }
198
199 static read_method_decl(filtered)
200 {
201         struct filtered_istream *fs = &(st->u.filtered);
202         size_t filled = 0;
203
204         while (sz) {
205                 /* do we already have filtered output? */
206                 if (fs->o_ptr < fs->o_end) {
207                         size_t to_move = fs->o_end - fs->o_ptr;
208                         if (sz < to_move)
209                                 to_move = sz;
210                         memcpy(buf + filled, fs->obuf + fs->o_ptr, to_move);
211                         fs->o_ptr += to_move;
212                         sz -= to_move;
213                         filled += to_move;
214                         continue;
215                 }
216                 fs->o_end = fs->o_ptr = 0;
217
218                 /* do we have anything to feed the filter with? */
219                 if (fs->i_ptr < fs->i_end) {
220                         size_t to_feed = fs->i_end - fs->i_ptr;
221                         size_t to_receive = FILTER_BUFFER;
222                         if (stream_filter(fs->filter,
223                                           fs->ibuf + fs->i_ptr, &to_feed,
224                                           fs->obuf, &to_receive))
225                                 return -1;
226                         fs->i_ptr = fs->i_end - to_feed;
227                         fs->o_end = FILTER_BUFFER - to_receive;
228                         continue;
229                 }
230
231                 /* tell the filter to drain upon no more input */
232                 if (fs->input_finished) {
233                         size_t to_receive = FILTER_BUFFER;
234                         if (stream_filter(fs->filter,
235                                           NULL, NULL,
236                                           fs->obuf, &to_receive))
237                                 return -1;
238                         fs->o_end = FILTER_BUFFER - to_receive;
239                         if (!fs->o_end)
240                                 break;
241                         continue;
242                 }
243                 fs->i_end = fs->i_ptr = 0;
244
245                 /* refill the input from the upstream */
246                 if (!fs->input_finished) {
247                         fs->i_end = read_istream(fs->upstream, fs->ibuf, FILTER_BUFFER);
248                         if (fs->i_end < 0)
249                                 return -1;
250                         if (fs->i_end)
251                                 continue;
252                 }
253                 fs->input_finished = 1;
254         }
255         return filled;
256 }
257
258 static struct stream_vtbl filtered_vtbl = {
259         close_istream_filtered,
260         read_istream_filtered,
261 };
262
263 static struct git_istream *attach_stream_filter(struct git_istream *st,
264                                                 struct stream_filter *filter)
265 {
266         struct git_istream *ifs = xmalloc(sizeof(*ifs));
267         struct filtered_istream *fs = &(ifs->u.filtered);
268
269         ifs->vtbl = &filtered_vtbl;
270         fs->upstream = st;
271         fs->filter = filter;
272         fs->i_end = fs->i_ptr = 0;
273         fs->o_end = fs->o_ptr = 0;
274         fs->input_finished = 0;
275         ifs->size = -1; /* unknown */
276         return ifs;
277 }
278
279 /*****************************************************************
280  *
281  * Loose object stream
282  *
283  *****************************************************************/
284
285 static read_method_decl(loose)
286 {
287         size_t total_read = 0;
288
289         switch (st->z_state) {
290         case z_done:
291                 return 0;
292         case z_error:
293                 return -1;
294         default:
295                 break;
296         }
297
298         if (st->u.loose.hdr_used < st->u.loose.hdr_avail) {
299                 size_t to_copy = st->u.loose.hdr_avail - st->u.loose.hdr_used;
300                 if (sz < to_copy)
301                         to_copy = sz;
302                 memcpy(buf, st->u.loose.hdr + st->u.loose.hdr_used, to_copy);
303                 st->u.loose.hdr_used += to_copy;
304                 total_read += to_copy;
305         }
306
307         while (total_read < sz) {
308                 int status;
309
310                 st->z.next_out = (unsigned char *)buf + total_read;
311                 st->z.avail_out = sz - total_read;
312                 status = git_inflate(&st->z, Z_FINISH);
313
314                 total_read = st->z.next_out - (unsigned char *)buf;
315
316                 if (status == Z_STREAM_END) {
317                         git_inflate_end(&st->z);
318                         st->z_state = z_done;
319                         break;
320                 }
321                 if (status != Z_OK && (status != Z_BUF_ERROR || total_read < sz)) {
322                         git_inflate_end(&st->z);
323                         st->z_state = z_error;
324                         return -1;
325                 }
326         }
327         return total_read;
328 }
329
330 static close_method_decl(loose)
331 {
332         close_deflated_stream(st);
333         munmap(st->u.loose.mapped, st->u.loose.mapsize);
334         return 0;
335 }
336
337 static struct stream_vtbl loose_vtbl = {
338         close_istream_loose,
339         read_istream_loose,
340 };
341
342 static open_method_decl(loose)
343 {
344         st->u.loose.mapped = map_loose_object(r, oid, &st->u.loose.mapsize);
345         if (!st->u.loose.mapped)
346                 return -1;
347         if ((unpack_loose_header(&st->z,
348                                  st->u.loose.mapped,
349                                  st->u.loose.mapsize,
350                                  st->u.loose.hdr,
351                                  sizeof(st->u.loose.hdr)) < 0) ||
352             (parse_loose_header(st->u.loose.hdr, &st->size) < 0)) {
353                 git_inflate_end(&st->z);
354                 munmap(st->u.loose.mapped, st->u.loose.mapsize);
355                 return -1;
356         }
357
358         st->u.loose.hdr_used = strlen(st->u.loose.hdr) + 1;
359         st->u.loose.hdr_avail = st->z.total_out;
360         st->z_state = z_used;
361
362         st->vtbl = &loose_vtbl;
363         return 0;
364 }
365
366
367 /*****************************************************************
368  *
369  * Non-delta packed object stream
370  *
371  *****************************************************************/
372
373 static read_method_decl(pack_non_delta)
374 {
375         size_t total_read = 0;
376
377         switch (st->z_state) {
378         case z_unused:
379                 memset(&st->z, 0, sizeof(st->z));
380                 git_inflate_init(&st->z);
381                 st->z_state = z_used;
382                 break;
383         case z_done:
384                 return 0;
385         case z_error:
386                 return -1;
387         case z_used:
388                 break;
389         }
390
391         while (total_read < sz) {
392                 int status;
393                 struct pack_window *window = NULL;
394                 unsigned char *mapped;
395
396                 mapped = use_pack(st->u.in_pack.pack, &window,
397                                   st->u.in_pack.pos, &st->z.avail_in);
398
399                 st->z.next_out = (unsigned char *)buf + total_read;
400                 st->z.avail_out = sz - total_read;
401                 st->z.next_in = mapped;
402                 status = git_inflate(&st->z, Z_FINISH);
403
404                 st->u.in_pack.pos += st->z.next_in - mapped;
405                 total_read = st->z.next_out - (unsigned char *)buf;
406                 unuse_pack(&window);
407
408                 if (status == Z_STREAM_END) {
409                         git_inflate_end(&st->z);
410                         st->z_state = z_done;
411                         break;
412                 }
413
414                 /*
415                  * Unlike the loose object case, we do not have to worry here
416                  * about running out of input bytes and spinning infinitely. If
417                  * we get Z_BUF_ERROR due to too few input bytes, then we'll
418                  * replenish them in the next use_pack() call when we loop. If
419                  * we truly hit the end of the pack (i.e., because it's corrupt
420                  * or truncated), then use_pack() catches that and will die().
421                  */
422                 if (status != Z_OK && status != Z_BUF_ERROR) {
423                         git_inflate_end(&st->z);
424                         st->z_state = z_error;
425                         return -1;
426                 }
427         }
428         return total_read;
429 }
430
431 static close_method_decl(pack_non_delta)
432 {
433         close_deflated_stream(st);
434         return 0;
435 }
436
437 static struct stream_vtbl pack_non_delta_vtbl = {
438         close_istream_pack_non_delta,
439         read_istream_pack_non_delta,
440 };
441
442 static open_method_decl(pack_non_delta)
443 {
444         struct pack_window *window;
445         enum object_type in_pack_type;
446
447         st->u.in_pack.pack = oi->u.packed.pack;
448         st->u.in_pack.pos = oi->u.packed.offset;
449         window = NULL;
450
451         in_pack_type = unpack_object_header(st->u.in_pack.pack,
452                                             &window,
453                                             &st->u.in_pack.pos,
454                                             &st->size);
455         unuse_pack(&window);
456         switch (in_pack_type) {
457         default:
458                 return -1; /* we do not do deltas for now */
459         case OBJ_COMMIT:
460         case OBJ_TREE:
461         case OBJ_BLOB:
462         case OBJ_TAG:
463                 break;
464         }
465         st->z_state = z_unused;
466         st->vtbl = &pack_non_delta_vtbl;
467         return 0;
468 }
469
470
471 /*****************************************************************
472  *
473  * In-core stream
474  *
475  *****************************************************************/
476
477 static close_method_decl(incore)
478 {
479         free(st->u.incore.buf);
480         return 0;
481 }
482
483 static read_method_decl(incore)
484 {
485         size_t read_size = sz;
486         size_t remainder = st->size - st->u.incore.read_ptr;
487
488         if (remainder <= read_size)
489                 read_size = remainder;
490         if (read_size) {
491                 memcpy(buf, st->u.incore.buf + st->u.incore.read_ptr, read_size);
492                 st->u.incore.read_ptr += read_size;
493         }
494         return read_size;
495 }
496
497 static struct stream_vtbl incore_vtbl = {
498         close_istream_incore,
499         read_istream_incore,
500 };
501
502 static open_method_decl(incore)
503 {
504         st->u.incore.buf = read_object_file_extended(r, oid, type, &st->size, 0);
505         st->u.incore.read_ptr = 0;
506         st->vtbl = &incore_vtbl;
507
508         return st->u.incore.buf ? 0 : -1;
509 }
510
511
512 /****************************************************************
513  * Users of streaming interface
514  ****************************************************************/
515
516 int stream_blob_to_fd(int fd, const struct object_id *oid, struct stream_filter *filter,
517                       int can_seek)
518 {
519         struct git_istream *st;
520         enum object_type type;
521         unsigned long sz;
522         ssize_t kept = 0;
523         int result = -1;
524
525         st = open_istream(the_repository, oid, &type, &sz, filter);
526         if (!st) {
527                 if (filter)
528                         free_stream_filter(filter);
529                 return result;
530         }
531         if (type != OBJ_BLOB)
532                 goto close_and_exit;
533         for (;;) {
534                 char buf[1024 * 16];
535                 ssize_t wrote, holeto;
536                 ssize_t readlen = read_istream(st, buf, sizeof(buf));
537
538                 if (readlen < 0)
539                         goto close_and_exit;
540                 if (!readlen)
541                         break;
542                 if (can_seek && sizeof(buf) == readlen) {
543                         for (holeto = 0; holeto < readlen; holeto++)
544                                 if (buf[holeto])
545                                         break;
546                         if (readlen == holeto) {
547                                 kept += holeto;
548                                 continue;
549                         }
550                 }
551
552                 if (kept && lseek(fd, kept, SEEK_CUR) == (off_t) -1)
553                         goto close_and_exit;
554                 else
555                         kept = 0;
556                 wrote = write_in_full(fd, buf, readlen);
557
558                 if (wrote < 0)
559                         goto close_and_exit;
560         }
561         if (kept && (lseek(fd, kept - 1, SEEK_CUR) == (off_t) -1 ||
562                      xwrite(fd, "", 1) != 1))
563                 goto close_and_exit;
564         result = 0;
565
566  close_and_exit:
567         close_istream(st);
568         return result;
569 }