streaming.c

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