granicus.if.org Git - postgresql/blob - src/backend/storage/file/buffile.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * buffile.c
   4  *        Management of large buffered files, primarily temporary files.
   5  *
   6  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  * IDENTIFICATION
  10  *        $PostgreSQL: pgsql/src/backend/storage/file/buffile.c,v 1.36 2010/01/02 16:57:51 momjian Exp $
  11  *
  12  * NOTES:
  13  *
  14  * BufFiles provide a very incomplete emulation of stdio atop virtual Files
  15  * (as managed by fd.c).  Currently, we only support the buffered-I/O
  16  * aspect of stdio: a read or write of the low-level File occurs only
  17  * when the buffer is filled or emptied.  This is an even bigger win
  18  * for virtual Files than for ordinary kernel files, since reducing the
  19  * frequency with which a virtual File is touched reduces "thrashing"
  20  * of opening/closing file descriptors.
  21  *
  22  * Note that BufFile structs are allocated with palloc(), and therefore
  23  * will go away automatically at transaction end.  If the underlying
  24  * virtual File is made with OpenTemporaryFile, then all resources for
  25  * the file are certain to be cleaned up even if processing is aborted
  26  * by ereport(ERROR).   To avoid confusion, the caller should take care that
  27  * all calls for a single BufFile are made in the same palloc context.
  28  *
  29  * BufFile also supports temporary files that exceed the OS file size limit
  30  * (by opening multiple fd.c temporary files).  This is an essential feature
  31  * for sorts and hashjoins on large amounts of data.
  32  *-------------------------------------------------------------------------
  33  */
  34
  35 #include "postgres.h"
  36
  37 #include "executor/instrument.h"
  38 #include "storage/fd.h"
  39 #include "storage/buffile.h"
  40 #include "storage/buf_internals.h"
  41
  42 /*
  43  * We break BufFiles into gigabyte-sized segments, regardless of RELSEG_SIZE.
  44  * The reason is that we'd like large temporary BufFiles to be spread across
  45  * multiple tablespaces when available.
  46  */
  47 #define MAX_PHYSICAL_FILESIZE   0x40000000
  48 #define BUFFILE_SEG_SIZE                (MAX_PHYSICAL_FILESIZE / BLCKSZ)
  49
  50 /*
  51  * This data structure represents a buffered file that consists of one or
  52  * more physical files (each accessed through a virtual file descriptor
  53  * managed by fd.c).
  54  */
  55 struct BufFile
  56 {
  57         int                     numFiles;               /* number of physical files in set */
  58         /* all files except the last have length exactly MAX_PHYSICAL_FILESIZE */
  59         File       *files;                      /* palloc'd array with numFiles entries */
  60         off_t      *offsets;            /* palloc'd array with numFiles entries */
  61
  62         /*
  63          * offsets[i] is the current seek position of files[i].  We use this to
  64          * avoid making redundant FileSeek calls.
  65          */
  66
  67         bool            isTemp;                 /* can only add files if this is TRUE */
  68         bool            isInterXact;    /* keep open over transactions? */
  69         bool            dirty;                  /* does buffer need to be written? */
  70
  71         /*
  72          * "current pos" is position of start of buffer within the logical file.
  73          * Position as seen by user of BufFile is (curFile, curOffset + pos).
  74          */
  75         int                     curFile;                /* file index (0..n) part of current pos */
  76         off_t           curOffset;              /* offset part of current pos */
  77         int                     pos;                    /* next read/write position in buffer */
  78         int                     nbytes;                 /* total # of valid bytes in buffer */
  79         char            buffer[BLCKSZ];
  80 };
  81
  82 static BufFile *makeBufFile(File firstfile);
  83 static void extendBufFile(BufFile *file);
  84 static void BufFileLoadBuffer(BufFile *file);
  85 static void BufFileDumpBuffer(BufFile *file);
  86 static int      BufFileFlush(BufFile *file);
  87
  88
  89 /*
  90  * Create a BufFile given the first underlying physical file.
  91  * NOTE: caller must set isTemp and isInterXact if appropriate.
  92  */
  93 static BufFile *
  94 makeBufFile(File firstfile)
  95 {
  96         BufFile    *file = (BufFile *) palloc(sizeof(BufFile));
  97
  98         file->numFiles = 1;
  99         file->files = (File *) palloc(sizeof(File));
 100         file->files[0] = firstfile;
 101         file->offsets = (off_t *) palloc(sizeof(off_t));
 102         file->offsets[0] = 0L;
 103         file->isTemp = false;
 104         file->isInterXact = false;
 105         file->dirty = false;
 106         file->curFile = 0;
 107         file->curOffset = 0L;
 108         file->pos = 0;
 109         file->nbytes = 0;
 110
 111         return file;
 112 }
 113
 114 /*
 115  * Add another component temp file.
 116  */
 117 static void
 118 extendBufFile(BufFile *file)
 119 {
 120         File            pfile;
 121
 122         Assert(file->isTemp);
 123         pfile = OpenTemporaryFile(file->isInterXact);
 124         Assert(pfile >= 0);
 125
 126         file->files = (File *) repalloc(file->files,
 127                                                                         (file->numFiles + 1) * sizeof(File));
 128         file->offsets = (off_t *) repalloc(file->offsets,
 129                                                                            (file->numFiles + 1) * sizeof(off_t));
 130         file->files[file->numFiles] = pfile;
 131         file->offsets[file->numFiles] = 0L;
 132         file->numFiles++;
 133 }
 134
 135 /*
 136  * Create a BufFile for a new temporary file (which will expand to become
 137  * multiple temporary files if more than MAX_PHYSICAL_FILESIZE bytes are
 138  * written to it).
 139  *
 140  * If interXact is true, the temp file will not be automatically deleted
 141  * at end of transaction.
 142  *
 143  * Note: if interXact is true, the caller had better be calling us in a
 144  * memory context that will survive across transaction boundaries.
 145  */
 146 BufFile *
 147 BufFileCreateTemp(bool interXact)
 148 {
 149         BufFile    *file;
 150         File            pfile;
 151
 152         pfile = OpenTemporaryFile(interXact);
 153         Assert(pfile >= 0);
 154
 155         file = makeBufFile(pfile);
 156         file->isTemp = true;
 157         file->isInterXact = interXact;
 158
 159         return file;
 160 }
 161
 162 #ifdef NOT_USED
 163 /*
 164  * Create a BufFile and attach it to an already-opened virtual File.
 165  *
 166  * This is comparable to fdopen() in stdio.  This is the only way at present
 167  * to attach a BufFile to a non-temporary file.  Note that BufFiles created
 168  * in this way CANNOT be expanded into multiple files.
 169  */
 170 BufFile *
 171 BufFileCreate(File file)
 172 {
 173         return makeBufFile(file);
 174 }
 175 #endif
 176
 177 /*
 178  * Close a BufFile
 179  *
 180  * Like fclose(), this also implicitly FileCloses the underlying File.
 181  */
 182 void
 183 BufFileClose(BufFile *file)
 184 {
 185         int                     i;
 186
 187         /* flush any unwritten data */
 188         BufFileFlush(file);
 189         /* close the underlying file(s) (with delete if it's a temp file) */
 190         for (i = 0; i < file->numFiles; i++)
 191                 FileClose(file->files[i]);
 192         /* release the buffer space */
 193         pfree(file->files);
 194         pfree(file->offsets);
 195         pfree(file);
 196 }
 197
 198 /*
 199  * BufFileLoadBuffer
 200  *
 201  * Load some data into buffer, if possible, starting from curOffset.
 202  * At call, must have dirty = false, pos and nbytes = 0.
 203  * On exit, nbytes is number of bytes loaded.
 204  */
 205 static void
 206 BufFileLoadBuffer(BufFile *file)
 207 {
 208         File            thisfile;
 209
 210         /*
 211          * Advance to next component file if necessary and possible.
 212          *
 213          * This path can only be taken if there is more than one component, so it
 214          * won't interfere with reading a non-temp file that is over
 215          * MAX_PHYSICAL_FILESIZE.
 216          */
 217         if (file->curOffset >= MAX_PHYSICAL_FILESIZE &&
 218                 file->curFile + 1 < file->numFiles)
 219         {
 220                 file->curFile++;
 221                 file->curOffset = 0L;
 222         }
 223
 224         /*
 225          * May need to reposition physical file.
 226          */
 227         thisfile = file->files[file->curFile];
 228         if (file->curOffset != file->offsets[file->curFile])
 229         {
 230                 if (FileSeek(thisfile, file->curOffset, SEEK_SET) != file->curOffset)
 231                         return;                         /* seek failed, read nothing */
 232                 file->offsets[file->curFile] = file->curOffset;
 233         }
 234
 235         /*
 236          * Read whatever we can get, up to a full bufferload.
 237          */
 238         file->nbytes = FileRead(thisfile, file->buffer, sizeof(file->buffer));
 239         if (file->nbytes < 0)
 240                 file->nbytes = 0;
 241         file->offsets[file->curFile] += file->nbytes;
 242         /* we choose not to advance curOffset here */
 243
 244         pgBufferUsage.temp_blks_read++;
 245 }
 246
 247 /*
 248  * BufFileDumpBuffer
 249  *
 250  * Dump buffer contents starting at curOffset.
 251  * At call, should have dirty = true, nbytes > 0.
 252  * On exit, dirty is cleared if successful write, and curOffset is advanced.
 253  */
 254 static void
 255 BufFileDumpBuffer(BufFile *file)
 256 {
 257         int                     wpos = 0;
 258         int                     bytestowrite;
 259         File            thisfile;
 260
 261         /*
 262          * Unlike BufFileLoadBuffer, we must dump the whole buffer even if it
 263          * crosses a component-file boundary; so we need a loop.
 264          */
 265         while (wpos < file->nbytes)
 266         {
 267                 /*
 268                  * Advance to next component file if necessary and possible.
 269                  */
 270                 if (file->curOffset >= MAX_PHYSICAL_FILESIZE && file->isTemp)
 271                 {
 272                         while (file->curFile + 1 >= file->numFiles)
 273                                 extendBufFile(file);
 274                         file->curFile++;
 275                         file->curOffset = 0L;
 276                 }
 277
 278                 /*
 279                  * Enforce per-file size limit only for temp files, else just try to
 280                  * write as much as asked...
 281                  */
 282                 bytestowrite = file->nbytes - wpos;
 283                 if (file->isTemp)
 284                 {
 285                         off_t           availbytes = MAX_PHYSICAL_FILESIZE - file->curOffset;
 286
 287                         if ((off_t) bytestowrite > availbytes)
 288                                 bytestowrite = (int) availbytes;
 289                 }
 290
 291                 /*
 292                  * May need to reposition physical file.
 293                  */
 294                 thisfile = file->files[file->curFile];
 295                 if (file->curOffset != file->offsets[file->curFile])
 296                 {
 297                         if (FileSeek(thisfile, file->curOffset, SEEK_SET) != file->curOffset)
 298                                 return;                 /* seek failed, give up */
 299                         file->offsets[file->curFile] = file->curOffset;
 300                 }
 301                 bytestowrite = FileWrite(thisfile, file->buffer + wpos, bytestowrite);
 302                 if (bytestowrite <= 0)
 303                         return;                         /* failed to write */
 304                 file->offsets[file->curFile] += bytestowrite;
 305                 file->curOffset += bytestowrite;
 306                 wpos += bytestowrite;
 307
 308                 pgBufferUsage.temp_blks_written++;
 309         }
 310         file->dirty = false;
 311
 312         /*
 313          * At this point, curOffset has been advanced to the end of the buffer,
 314          * ie, its original value + nbytes.  We need to make it point to the
 315          * logical file position, ie, original value + pos, in case that is less
 316          * (as could happen due to a small backwards seek in a dirty buffer!)
 317          */
 318         file->curOffset -= (file->nbytes - file->pos);
 319         if (file->curOffset < 0)        /* handle possible segment crossing */
 320         {
 321                 file->curFile--;
 322                 Assert(file->curFile >= 0);
 323                 file->curOffset += MAX_PHYSICAL_FILESIZE;
 324         }
 325
 326         /*
 327          * Now we can set the buffer empty without changing the logical position
 328          */
 329         file->pos = 0;
 330         file->nbytes = 0;
 331 }
 332
 333 /*
 334  * BufFileRead
 335  *
 336  * Like fread() except we assume 1-byte element size.
 337  */
 338 size_t
 339 BufFileRead(BufFile *file, void *ptr, size_t size)
 340 {
 341         size_t          nread = 0;
 342         size_t          nthistime;
 343
 344         if (file->dirty)
 345         {
 346                 if (BufFileFlush(file) != 0)
 347                         return 0;                       /* could not flush... */
 348                 Assert(!file->dirty);
 349         }
 350
 351         while (size > 0)
 352         {
 353                 if (file->pos >= file->nbytes)
 354                 {
 355                         /* Try to load more data into buffer. */
 356                         file->curOffset += file->pos;
 357                         file->pos = 0;
 358                         file->nbytes = 0;
 359                         BufFileLoadBuffer(file);
 360                         if (file->nbytes <= 0)
 361                                 break;                  /* no more data available */
 362                 }
 363
 364                 nthistime = file->nbytes - file->pos;
 365                 if (nthistime > size)
 366                         nthistime = size;
 367                 Assert(nthistime > 0);
 368
 369                 memcpy(ptr, file->buffer + file->pos, nthistime);
 370
 371                 file->pos += nthistime;
 372                 ptr = (void *) ((char *) ptr + nthistime);
 373                 size -= nthistime;
 374                 nread += nthistime;
 375         }
 376
 377         return nread;
 378 }
 379
 380 /*
 381  * BufFileWrite
 382  *
 383  * Like fwrite() except we assume 1-byte element size.
 384  */
 385 size_t
 386 BufFileWrite(BufFile *file, void *ptr, size_t size)
 387 {
 388         size_t          nwritten = 0;
 389         size_t          nthistime;
 390
 391         while (size > 0)
 392         {
 393                 if (file->pos >= BLCKSZ)
 394                 {
 395                         /* Buffer full, dump it out */
 396                         if (file->dirty)
 397                         {
 398                                 BufFileDumpBuffer(file);
 399                                 if (file->dirty)
 400                                         break;          /* I/O error */
 401                         }
 402                         else
 403                         {
 404                                 /* Hmm, went directly from reading to writing? */
 405                                 file->curOffset += file->pos;
 406                                 file->pos = 0;
 407                                 file->nbytes = 0;
 408                         }
 409                 }
 410
 411                 nthistime = BLCKSZ - file->pos;
 412                 if (nthistime > size)
 413                         nthistime = size;
 414                 Assert(nthistime > 0);
 415
 416                 memcpy(file->buffer + file->pos, ptr, nthistime);
 417
 418                 file->dirty = true;
 419                 file->pos += nthistime;
 420                 if (file->nbytes < file->pos)
 421                         file->nbytes = file->pos;
 422                 ptr = (void *) ((char *) ptr + nthistime);
 423                 size -= nthistime;
 424                 nwritten += nthistime;
 425         }
 426
 427         return nwritten;
 428 }
 429
 430 /*
 431  * BufFileFlush
 432  *
 433  * Like fflush()
 434  */
 435 static int
 436 BufFileFlush(BufFile *file)
 437 {
 438         if (file->dirty)
 439         {
 440                 BufFileDumpBuffer(file);
 441                 if (file->dirty)
 442                         return EOF;
 443         }
 444
 445         return 0;
 446 }
 447
 448 /*
 449  * BufFileSeek
 450  *
 451  * Like fseek(), except that target position needs two values in order to
 452  * work when logical filesize exceeds maximum value representable by long.
 453  * We do not support relative seeks across more than LONG_MAX, however.
 454  *
 455  * Result is 0 if OK, EOF if not.  Logical position is not moved if an
 456  * impossible seek is attempted.
 457  */
 458 int
 459 BufFileSeek(BufFile *file, int fileno, off_t offset, int whence)
 460 {
 461         int                     newFile;
 462         off_t           newOffset;
 463
 464         switch (whence)
 465         {
 466                 case SEEK_SET:
 467                         if (fileno < 0)
 468                                 return EOF;
 469                         newFile = fileno;
 470                         newOffset = offset;
 471                         break;
 472                 case SEEK_CUR:
 473
 474                         /*
 475                          * Relative seek considers only the signed offset, ignoring
 476                          * fileno. Note that large offsets (> 1 gig) risk overflow in this
 477                          * add, unless we have 64-bit off_t.
 478                          */
 479                         newFile = file->curFile;
 480                         newOffset = (file->curOffset + file->pos) + offset;
 481                         break;
 482 #ifdef NOT_USED
 483                 case SEEK_END:
 484                         /* could be implemented, not needed currently */
 485                         break;
 486 #endif
 487                 default:
 488                         elog(ERROR, "invalid whence: %d", whence);
 489                         return EOF;
 490         }
 491         while (newOffset < 0)
 492         {
 493                 if (--newFile < 0)
 494                         return EOF;
 495                 newOffset += MAX_PHYSICAL_FILESIZE;
 496         }
 497         if (newFile == file->curFile &&
 498                 newOffset >= file->curOffset &&
 499                 newOffset <= file->curOffset + file->nbytes)
 500         {
 501                 /*
 502                  * Seek is to a point within existing buffer; we can just adjust
 503                  * pos-within-buffer, without flushing buffer.  Note this is OK
 504                  * whether reading or writing, but buffer remains dirty if we were
 505                  * writing.
 506                  */
 507                 file->pos = (int) (newOffset - file->curOffset);
 508                 return 0;
 509         }
 510         /* Otherwise, must reposition buffer, so flush any dirty data */
 511         if (BufFileFlush(file) != 0)
 512                 return EOF;
 513
 514         /*
 515          * At this point and no sooner, check for seek past last segment. The
 516          * above flush could have created a new segment, so checking sooner would
 517          * not work (at least not with this code).
 518          */
 519         if (file->isTemp)
 520         {
 521                 /* convert seek to "start of next seg" to "end of last seg" */
 522                 if (newFile == file->numFiles && newOffset == 0)
 523                 {
 524                         newFile--;
 525                         newOffset = MAX_PHYSICAL_FILESIZE;
 526                 }
 527                 while (newOffset > MAX_PHYSICAL_FILESIZE)
 528                 {
 529                         if (++newFile >= file->numFiles)
 530                                 return EOF;
 531                         newOffset -= MAX_PHYSICAL_FILESIZE;
 532                 }
 533         }
 534         if (newFile >= file->numFiles)
 535                 return EOF;
 536         /* Seek is OK! */
 537         file->curFile = newFile;
 538         file->curOffset = newOffset;
 539         file->pos = 0;
 540         file->nbytes = 0;
 541         return 0;
 542 }
 543
 544 void
 545 BufFileTell(BufFile *file, int *fileno, off_t *offset)
 546 {
 547         *fileno = file->curFile;
 548         *offset = file->curOffset + file->pos;
 549 }
 550
 551 /*
 552  * BufFileSeekBlock --- block-oriented seek
 553  *
 554  * Performs absolute seek to the start of the n'th BLCKSZ-sized block of
 555  * the file.  Note that users of this interface will fail if their files
 556  * exceed BLCKSZ * LONG_MAX bytes, but that is quite a lot; we don't work
 557  * with tables bigger than that, either...
 558  *
 559  * Result is 0 if OK, EOF if not.  Logical position is not moved if an
 560  * impossible seek is attempted.
 561  */
 562 int
 563 BufFileSeekBlock(BufFile *file, long blknum)
 564 {
 565         return BufFileSeek(file,
 566                                            (int) (blknum / BUFFILE_SEG_SIZE),
 567                                            (off_t) (blknum % BUFFILE_SEG_SIZE) * BLCKSZ,
 568                                            SEEK_SET);
 569 }
 570
 571 #ifdef NOT_USED
 572 /*
 573  * BufFileTellBlock --- block-oriented tell
 574  *
 575  * Any fractional part of a block in the current seek position is ignored.
 576  */
 577 long
 578 BufFileTellBlock(BufFile *file)
 579 {
 580         long            blknum;
 581
 582         blknum = (file->curOffset + file->pos) / BLCKSZ;
 583         blknum += file->curFile * BUFFILE_SEG_SIZE;
 584         return blknum;
 585 }
 586
 587 #endif