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