dwbuf2.patch

text/x-patch

Download

Patch

Same data as JSON: GET /api/v1/attachments/:id/patch the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes. API reference →

Format: unified

File	+	−
src/backend/access/heap/heapam.c	7	3
src/backend/access/heap/rewriteheap.c	1	1
src/backend/access/nbtree/nbtree.c	1	1
src/backend/access/nbtree/nbtsort.c	1	1
src/backend/access/spgist/spginsert.c	2	2
src/backend/access/transam/xlog.c	13	0
src/backend/commands/copy.c	1	1
src/backend/commands/tablecmds.c	3	3
src/backend/executor/execMain.c	1	1
src/backend/postmaster/checkpointer.c	9	0
src/backend/postmaster/postmaster.c	8	0
src/backend/storage/buffer/buf_init.c	39	1
src/backend/storage/buffer/bufmgr.c	90	21
src/backend/storage/buffer/ioseq.c	250	0
src/backend/storage/buffer/localbuf.c	1	1
src/backend/storage/buffer/Makefile	1	1
src/backend/storage/buffer/README.doublewrites	186	0
src/backend/storage/file/fd.c	229	1
src/backend/storage/ipc/ipci.c	4	1
src/backend/storage/lmgr/lwlock.c	7	1
src/backend/storage/smgr/md.c	49	0
src/backend/storage/smgr/smgr.c	866	6
src/backend/utils/misc/guc.c	53	0
src/backend/utils/misc/postgresql.conf.sample	4	0
src/include/access/heapam.h	1	1
src/include/postmaster/bgwriter.h	2	0
src/include/storage/bufmgr.h	1	1
src/include/storage/bufpage.h	3	1
src/include/storage/fd.h	9	0
src/include/storage/ioseq.h	37	0
src/include/storage/smgr.h	123	2

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 99a431a..37da9d0 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -5722,16 +5722,20 @@ heap2_desc(StringInfo buf, uint8 xl_info, char *rec)
  * That behavior might change someday, but in any case it's likely that
  * any fsync decisions required would be per-index and hence not appropriate
  * to be done here.)
+ *
+ * If needsDoubleWrite is true, the changed buffers should use
+ * double-writing if that option is enabled.  Currently, relation changes
+ * that were not WAL-logged do not need double writes.
  */
 void
-heap_sync(Relation rel)
+heap_sync(Relation rel, bool needsDoubleWrite)
 {
 	/* non-WAL-logged tables never need fsync */
 	if (!RelationNeedsWAL(rel))
 		return;
 
 	/* main heap */
-	FlushRelationBuffers(rel);
+	FlushRelationBuffers(rel, needsDoubleWrite);
 	/* FlushRelationBuffers will have opened rd_smgr */
 	smgrimmedsync(rel->rd_smgr, MAIN_FORKNUM);
 
@@ -5743,7 +5747,7 @@ heap_sync(Relation rel)
 		Relation	toastrel;
 
 		toastrel = heap_open(rel->rd_rel->reltoastrelid, AccessShareLock);
-		FlushRelationBuffers(toastrel);
+		FlushRelationBuffers(toastrel, needsDoubleWrite);
 		smgrimmedsync(toastrel->rd_smgr, MAIN_FORKNUM);
 		heap_close(toastrel, AccessShareLock);
 	}
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index 31b2b67..9e44629 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -288,7 +288,7 @@ end_heap_rewrite(RewriteState state)
 	 * wrote before the checkpoint.
 	 */
 	if (RelationNeedsWAL(state->rs_new_rel))
-		heap_sync(state->rs_new_rel);
+		heap_sync(state->rs_new_rel, state->rs_use_wal);
 
 	/* Deleting the context frees everything */
 	MemoryContextDelete(state->rs_cxt);
diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index 7627738..f628545 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -217,7 +217,7 @@ btbuildempty(PG_FUNCTION_ARGS)
 
 	/* Write the page.	If archiving/streaming, XLOG it. */
 	smgrwrite(index->rd_smgr, INIT_FORKNUM, BTREE_METAPAGE,
-			  (char *) metapage, true);
+			  (char *) metapage, true, false);
 	if (XLogIsNeeded())
 		log_newpage(&index->rd_smgr->smgr_rnode.node, INIT_FORKNUM,
 					BTREE_METAPAGE, metapage);
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 9aa3a13..b7bb0d3 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -309,7 +309,7 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	{
 		/* overwriting a block we zero-filled before */
 		smgrwrite(wstate->index->rd_smgr, MAIN_FORKNUM, blkno,
-				  (char *) page, true);
+				  (char *) page, true, false);
 	}
 
 	pfree(page);
diff --git a/src/backend/access/spgist/spginsert.c b/src/backend/access/spgist/spginsert.c
index cbcf655..550b2cd 100644
--- a/src/backend/access/spgist/spginsert.c
+++ b/src/backend/access/spgist/spginsert.c
@@ -151,7 +151,7 @@ spgbuildempty(PG_FUNCTION_ARGS)
 
 	/* Write the page.	If archiving/streaming, XLOG it. */
 	smgrwrite(index->rd_smgr, INIT_FORKNUM, SPGIST_METAPAGE_BLKNO,
-			  (char *) page, true);
+			  (char *) page, true, false);
 	if (XLogIsNeeded())
 		log_newpage(&index->rd_smgr->smgr_rnode.node, INIT_FORKNUM,
 					SPGIST_METAPAGE_BLKNO, page);
@@ -160,7 +160,7 @@ spgbuildempty(PG_FUNCTION_ARGS)
 	SpGistInitPage(page, SPGIST_LEAF);
 
 	smgrwrite(index->rd_smgr, INIT_FORKNUM, SPGIST_HEAD_BLKNO,
-			  (char *) page, true);
+			  (char *) page, true, false);
 	if (XLogIsNeeded())
 		log_newpage(&index->rd_smgr->smgr_rnode.node, INIT_FORKNUM,
 					SPGIST_HEAD_BLKNO, page);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 4b273a8..82ab792 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6294,6 +6294,14 @@ StartupXLOG(void)
 		InRecovery = true;
 	}
 
+	/*
+	 * If double write file exists, see if there are any pages to be recovered
+	 * because of torn writes.	This must be done whether or not double_writes
+	 * or full_page_writes is currently enabled, in order to recover any torn
+	 * pages if double_writes was enabled during last crash.
+	 */
+	RecoverDoubleWriteFile();
+
 	/* REDO */
 	if (InRecovery)
 	{
@@ -7483,6 +7491,11 @@ ShutdownXLOG(int code, Datum arg)
 
 		CreateCheckPoint(CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_IMMEDIATE);
 	}
+
+	/* Flush double-write buffer before shutting down */
+	FlushDoubleWriteBuffer(DWBUF_NON_CHECKPOINTER);
+	FlushDoubleWriteBuffer(DWBUF_CHECKPOINTER);
+
 	ShutdownCLOG();
 	ShutdownSUBTRANS();
 	ShutdownMultiXact();
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 110480f..17c9b81 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -2138,7 +2138,7 @@ CopyFrom(CopyState cstate)
 	 * indexes since those use WAL anyway)
 	 */
 	if (hi_options & HEAP_INSERT_SKIP_WAL)
-		heap_sync(cstate->rel);
+		heap_sync(cstate->rel, false);
 
 	return processed;
 }
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 07dc326..9ea03a8 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -3771,7 +3771,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
 
 		/* If we skipped writing WAL, then we need to sync the heap. */
 		if (hi_options & HEAP_INSERT_SKIP_WAL)
-			heap_sync(newrel);
+			heap_sync(newrel, false);
 
 		heap_close(newrel, NoLock);
 	}
@@ -8433,7 +8433,7 @@ ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode)
 	 * in shared buffers.  We assume no new changes will be made while we are
 	 * holding exclusive lock on the rel.
 	 */
-	FlushRelationBuffers(rel);
+	FlushRelationBuffers(rel, true);
 
 	/*
 	 * Relfilenodes are not unique across tablespaces, so we need to allocate
@@ -8535,7 +8535,7 @@ copy_relation_data(SMgrRelation src, SMgrRelation dst,
 		/* If we got a cancel signal during the copy of the data, quit */
 		CHECK_FOR_INTERRUPTS();
 
-		smgrread(src, forkNum, blkno, buf);
+		smgrread(src, forkNum, blkno, buf, NULL);
 
 		/* XLOG stuff */
 		if (use_wal)
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 422f737..0b3742b 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -2692,7 +2692,7 @@ CloseIntoRel(QueryDesc *queryDesc)
 
 		/* If we skipped using WAL, must heap_sync before commit */
 		if (myState->hi_options & HEAP_INSERT_SKIP_WAL)
-			heap_sync(myState->rel);
+			heap_sync(myState->rel, false);
 
 		/* close rel, but keep lock until commit */
 		heap_close(myState->rel, NoLock);
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index 76cb25c..bb494f2 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -1257,6 +1257,15 @@ AbsorbFsyncRequests(void)
 }
 
 /*
+ * Return TRUE if the current process is the checkpointer.
+ */
+bool
+AmCheckpointer()
+{
+	return am_checkpointer;
+}
+
+/*
  * Update any shared memory configurations based on config parameters
  */
 static void
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 9d242cb..58bac88 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -314,6 +314,9 @@ extern char *optarg;
 extern int	optind,
 			opterr;
 
+extern int	page_checksum;
+extern bool	doubleWrites;
+
 #ifdef HAVE_INT_OPTRESET
 extern int	optreset;			/* might not be declared by system headers */
 #endif
@@ -765,6 +768,11 @@ PostmasterMain(int argc, char *argv[])
 		ereport(ERROR,
 				(errmsg("WAL streaming (max_wal_senders > 0) requires wal_level \"archive\" or \"hot_standby\"")));
 
+	/* The double-write option currently requires data page checksums. */
+	if (doubleWrites && !page_checksum)
+		ereport(ERROR,
+				(errmsg("page_checksum must be enabled if double_writes is enabled")));
+
 	/*
 	 * Other one-time internal sanity checks can go here, if they are fast.
 	 * (Put any slow processing further down, after postmaster.pid creation.)
diff --git a/src/backend/storage/buffer/Makefile b/src/backend/storage/buffer/Makefile
index 2c10fba..eb7f9c9 100644
--- a/src/backend/storage/buffer/Makefile
+++ b/src/backend/storage/buffer/Makefile
@@ -12,6 +12,6 @@ subdir = src/backend/storage/buffer
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = buf_table.o buf_init.o bufmgr.o freelist.o localbuf.o
+OBJS = buf_table.o buf_init.o bufmgr.o freelist.o localbuf.o ioseq.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/buffer/README.doublewrites b/src/backend/storage/buffer/README.doublewrites
new file mode 100644
index 0000000..d82af8e
--- /dev/null
+++ b/src/backend/storage/buffer/README.doublewrites
@@ -0,0 +1,186 @@
+DOUBLE WRITES
+
+The "double_write" GUC option enables the use of double write
+functionality, which can be used as a replacement for full page writes.
+The idea of this option is to handle the problem of torn writes for
+buffer pages by writing (almost) all buffers twice, once to a
+double-write file and once to the data file.  The two writes are done in
+a strictly sequential order, ensuring that the buffer is successfully
+written to the double-write file before being written to the data file.
+The "double_write" option (like full page writes) is only a crash
+recovery feature -- it does not affect the database contents unless there
+is a crash.
+
+The double-write file has checksummed contents, and in our
+implementation, the pages in the data files are also be checksummed,
+though this is not strictly required.  If a crash occurs while buffer
+writes (using double writes) are in progress, then a buffer page may have
+a torn page in either the double-write file or in its data file.
+However, because of the ordering of writes, there can only be a torn page
+for that buffer in the double-write file or the data file, not both.
+Therefore, during crash recovery, we can scan the double-write file.  If
+a page in the double-write file has a correct checksum, and the
+corresponding page in the data file has an incorrect checksum, then we
+can restore the torn data page from the double-write file.  Any pages in
+the double-write file that have incorrect checksum are ignored (since
+they are likely torn pages).
+
+The net result is that "double_write" option fixes all torn data pages,
+and can therefore be used in place of "full_page_writes".  Using
+double_writes may improve performance, as compared to
+"full_page_writes", and always greatly reduces the size of the WAL log.
+
+As currently written, the double_write option makes use of checksums on
+the data pages.  Double writes only strictly require that the pages in
+the double-write file be checksummed, and we could fairly easily change
+the implementation to make data checksums optional.  However, if data
+checksums are used, then Postgres can provide more useful messages on
+exactly when torn pages have occurred.  It is very likely that a torn
+page happened if, during recovery, the checksum of a data page is
+incorrect, but a copy of the page with a valid checksum is in the
+double-write file.  If there are no data checksums, then Postgres would
+would still copy any valid page in the double-write file to the
+appropriate page in a data file, but it cannot actually know if a torn
+page occurred.
+
+
+BATCHING WRITES USING A DOUBLE-WRITE BUFFER
+
+A double write requires that data be written first to the
+double-write file and fully committed to disk, and then to the data
+files.  The data files must also be fully committed to disk so that the
+double-write file can be re-used.  Because of the two writes and the
+associated fsyncs, double writes can be expensive.  One way to improve
+the efficiency of double writes is to do double writes for multiple pages
+at once.
+
+The current implementation does this batching of page writes by using a
+"double-write buffer".  For each page that the checkpointer, bgwriter,
+autovacuum process, or backend need to write out with double writes, the
+process puts the page (sequentially) in a double-write buffer and just
+continues on.  Note that the page must be fully copied into the
+double-write buffer (which can be folded in with a checksum computation
+if enabled), since the process will release locks and continue without
+waiting.  When the double-write buffer is full, the process that made it
+full then flushes the buffer to disk, doing the appropriate writes and
+fsyncs to the double-write file and then to the data files.
+
+Since the pages in the double-write buffer are "on the way" to disk, the
+double-write buffer must checked first when processes go to read pages
+from disk.  Conveniently, this lookup can be done at the beginning of
+smgrread().  The copying of a page to the double-write buffer
+(instead of going immediately to disk) can be done at the beginning of
+smgrwrite() if double writes are enabled.
+
+Another important point is that the current contents of the double-write
+buffer must be flushed as a checkpoint completes to ensure the contents
+of the checkpoint is on disk.
+
+To reduce blocking when one process is actually writing out the contents
+of the double-write buffer, we have actually divided the double-write
+buffer into batches.  For example, the double-write buffer could be 128
+pages long, but with batches of 32 each.  When any process completes a
+batch as it adds its page to the double-write buffer, that process then
+flushes the batch to disk (doing the appropriate double write).  The
+double-write buffer is a circular buffer that is reused appropriately.
+To avoid extra contention, each batch in the double-write buffer writes
+to a different double-write file, so each batch can be written out
+independently as soon as the batch is filled.
+
+Our design also has two independent double-write buffers: one for the
+checkpointer and one for all other processes.  The checkpointer gets its
+own buffer in order to help ensure that it complete its checkpoints on
+time.  We have also included some code (in ioseq.h/ioseq.c) that sorts
+buffers to be checkpointed in file/block order.  This sorting greatly
+improves the performance of double-writes for the checkpointer, because
+it ensures that each batch writes to only one or a few data files.  This
+minimizes the number of fsyncs to data files that must be done by each
+batch double write.  In addition, the sorting makes it more likely that
+writes to sequential locations in a data file can be merged.  The use of
+a separate double-write buffer for the checkpointer ensures that the
+benefit of the sorting is maximized.
+
+The actual batch writes are done using writev(), which might have to be
+replaced with equivalent code, if this is a portability issue.  A struct
+iocb structure is currently used for bookkeeping during the low-level
+batching, since it is compatible with an async IO approach as well (not
+included). Given the batching functionality, double writes are
+implemented efficiently by writing a batch of pages to the double-write
+file and fsyncing, and then writing the pages to the appropriate data
+files, and then fsyncing all the necessary data files.  While the data
+fsyncing might be viewed as expensive, it does help eliminate a lot of
+the fsync overhead at the end of checkpoints.
+
+Note that direct IO (the O_DIRECT option in Linux) could be used for the
+double-write file, if there were no portability concerns.  In that case,
+we write the whole set of buffers to the double-write file in a single
+direct write, and no fsync in necessary.  We would avoid copying
+overhead, and any use of the buffer cache.  We would in that case need to
+make sure that the double-write file is fully-allocated ahead of time
+(which is the only case when Linux actually guarantees that IO will be
+direct to disk).  Or we could do the fsync just to guarantee that that
+the IO makes it to disk.
+
+The fsyncing of the data files is necessary to be sure that the
+double-write file can be reused for the next batch of buffers.  All
+buffers must be successfully written to the data files on disk before the
+double-write file is reused.  Strictly speaking, the data files only need
+to be fsync'ed by the time that the double-write file must be used again.
+
+
+DOUBLE-WRITE FILES
+
+When we write to a double-write file, we include a header that identifies
+each block in the double-write file and a checksum for each block (and is
+checksummed itself).  We use this header during recovery to determine the
+contents of the double-write file and whether each block was fully
+written (i.e. not torn).  The length of the double-write header is set as
+4096 bytes, and must be large enough to hold information on
+MAX_BATCH_BLOCKS blocks.  (See Assert() in smgrbwrite().)
+
+
+WHEN TO DO DOUBLE WRITES
+
+Double writes must be done for any page which might be used after
+recovery even if there was a full crash while writing the page.  This
+includes all writes to such pages in a checkpoint, not just the first
+write.  Pages in temporary tables and some unlogged operations do not
+require double writes.  This is controlled by flags in heap_sync() and
+FlushRelationBuffers().  The double writes must be done whether the
+process doing the writing is a checkpointer, bgwriter, vacuumer, or
+backend.  Fortunately, the double-write buffer helps merge together
+double-writes by individual backends.
+
+
+PERFORMANCE
+
+We have seen significant performance gains for OLTP runs with sufficient
+buffer cache size.  In these runs, there are smaller numbers of dirty
+evictions, and the checkpointer is doing the majority of the buffer writes.
+In that case, the number of writes is not being greatly increased.  We
+are doing each buffer write twice, but we are eliminating the full page
+write that goes to WAL log for each modification of a buffer in a
+checkpoint.  Also, the double-write file is written sequentially in a
+single write, so it is highly efficient.  Though we are doing many
+fsyncs, we are doing them in the context of the checkpointer (and
+possibly in the bgwriter).  Meanwhile, we have removed the latency added
+to each transaction when the backend must sync a full page write to the WAL
+log for the first modification of a buffer.
+
+OPTIONS
+
+The three options in this patch are double_writes,
+double_write_directory, and batched_buffer_writes.
+
+double_writes controls whether double writes are used for buffer writes.
+It requires that page_checksum is enabled, and that batched_buffer_writes
+is non-zero.  Generally, the user would turn off full_page_writes is
+double_writes is enabled.
+
+double_write_directory controls the directory where the double-write file
+is located.  If it is not set, then the double-write file is in the base
+directory.  This option could be used to put the double-write file on some
+high-speed media (such as SSD).
+
+The batched_buffer_writes option controls the batch size and must divide
+evenly into the double-write buffer sizes.
diff --git a/src/backend/storage/buffer/buf_init.c b/src/backend/storage/buffer/buf_init.c
index 94cefba..eab58e4 100644
--- a/src/backend/storage/buffer/buf_init.c
+++ b/src/backend/storage/buffer/buf_init.c
@@ -74,6 +74,7 @@ InitBufferPool(void)
 {
 	bool		foundBufs,
 				foundDescs;
+	int			i;
 
 	BufferDescriptors = (BufferDesc *)
 		ShmemInitStruct("Buffer Descriptors",
@@ -92,7 +93,6 @@ InitBufferPool(void)
 	else
 	{
 		BufferDesc *buf;
-		int			i;
 
 		buf = BufferDescriptors;
 
@@ -125,6 +125,44 @@ InitBufferPool(void)
 		BufferDescriptors[NBuffers - 1].freeNext = FREENEXT_END_OF_LIST;
 	}
 
+	/*
+	 * Allocate shared region for the double-write buffers and info on the
+	 * state of the buffers.
+	 */
+	dwBlocks = (char *)
+		ShmemInitStruct("DW Buffer",
+						(NON_CKPT_DW_BLOCKS + CKPT_DW_BLOCKS) * BLCKSZ + BLCKSZ,
+						&foundBufs);
+	Assert(!foundBufs);
+	dwBlocks = (char *) TYPEALIGN(BLCKSZ, dwBlocks);
+
+	dwInfo = (struct DWBufferInfo *)
+		ShmemInitStruct("DW Info", 2 * sizeof(struct DWBufferInfo), &foundBufs);
+	Assert(!foundBufs);
+
+	dwInfo[DWBUF_NON_CHECKPOINTER].bufferStart = dwBlocks;
+	dwInfo[DWBUF_NON_CHECKPOINTER].writeLock = LWLockAssign();
+	dwInfo[DWBUF_NON_CHECKPOINTER].allocLock = LWLockAssign();
+	dwInfo[DWBUF_NON_CHECKPOINTER].length = NON_CKPT_DW_BLOCKS;
+	dwInfo[DWBUF_NON_CHECKPOINTER].mask = NON_CKPT_DW_BLOCKS - 1;
+	Assert(dwInfo[DWBUF_NON_CHECKPOINTER].length % batched_buffer_writes == 0);
+	/* Create one lock per batch */
+	for (i = 0; i < dwInfo[DWBUF_NON_CHECKPOINTER].length;
+		 i += batched_buffer_writes) {
+		dwInfo[DWBUF_NON_CHECKPOINTER].batchLocks[i] = LWLockAssign();
+	}
+
+	dwInfo[DWBUF_CHECKPOINTER].bufferStart = dwBlocks + dwInfo[DWBUF_NON_CHECKPOINTER].length * BLCKSZ;
+	dwInfo[DWBUF_CHECKPOINTER].writeLock = LWLockAssign();
+	dwInfo[DWBUF_CHECKPOINTER].length = CKPT_DW_BLOCKS;
+	dwInfo[DWBUF_CHECKPOINTER].mask = CKPT_DW_BLOCKS - 1;
+	Assert(dwInfo[DWBUF_CHECKPOINTER].length % batched_buffer_writes == 0);
+	/* Create one lock per batch */
+	for (i = 0; i < dwInfo[DWBUF_CHECKPOINTER].length;
+		 i += batched_buffer_writes) {
+		dwInfo[DWBUF_CHECKPOINTER].batchLocks[i] = LWLockAssign();
+	}
+
 	/* Init other shared buffer-management stuff */
 	StrategyInitialize(!foundDescs);
 }
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 1adb6d3..d34791b 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -44,6 +44,7 @@
 #include "storage/ipc.h"
 #include "storage/proc.h"
 #include "storage/smgr.h"
+#include "storage/ioseq.h"
 #include "storage/standby.h"
 #include "utils/rel.h"
 #include "utils/resowner.h"
@@ -104,7 +105,8 @@ static volatile BufferDesc *BufferAlloc(SMgrRelation smgr,
 			BlockNumber blockNum,
 			BufferAccessStrategy strategy,
 			bool *foundPtr);
-static void FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln);
+static void FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln,
+	bool needsDoubleWrite);
 static void AtProcExit_Buffers(int code, Datum arg);
 
 
@@ -437,7 +439,7 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 			MemSet((char *) bufBlock, 0, BLCKSZ);
 		else
 		{
-			smgrread(smgr, forkNum, blockNum, (char *) bufBlock);
+			smgrread(smgr, forkNum, blockNum, (char *) bufBlock, NULL);
 
 			/* check for garbage data */
 			if (!PageHeaderIsValid((PageHeader) bufBlock))
@@ -651,7 +653,7 @@ BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 												smgr->smgr_rnode.node.dbNode,
 											  smgr->smgr_rnode.node.relNode);
 
-				FlushBuffer(buf, NULL);
+				FlushBuffer(buf, NULL, true);
 				LWLockRelease(buf->content_lock);
 
 				TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(forkNum, blockNum,
@@ -1185,11 +1187,19 @@ static void
 BufferSync(int flags)
 {
 	int			buf_id;
-	int			num_to_scan;
 	int			num_to_write;
 	int			num_written;
+	unsigned	i, seqIdx;
 	int			mask = BM_DIRTY;
 
+#define DBLWRITE_DEBUG
+#ifdef DBLWRITE_DEBUG
+	int64		targetTime,
+				now,
+				start;
+	int			cur_num_written;
+#endif
+
 	/* Make sure we can handle the pin inside SyncOneBuffer */
 	ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
 
@@ -1230,6 +1240,8 @@ BufferSync(int flags)
 		if ((bufHdr->flags & mask) == mask)
 		{
 			bufHdr->flags |= BM_CHECKPOINT_NEEDED;
+			ioSeqList[num_to_write].idx = buf_id;
+			ioSeqList[num_to_write].tag = bufHdr->tag;
 			num_to_write++;
 		}
 
@@ -1239,23 +1251,41 @@ BufferSync(int flags)
 	if (num_to_write == 0)
 		return;					/* nothing to do */
 
+	if (sortDirtyBuffers)
+	{
+		IOSeq_Sort(ioSeqList, num_to_write);
+		seqIdx = 0;
+	}
+	else
+	{
+		seqIdx = IOSeq_ClockStart(ioSeqList, num_to_write);
+	}
 	TRACE_POSTGRESQL_BUFFER_SYNC_START(NBuffers, num_to_write);
 
+#ifdef DBLWRITE_DEBUG
+	if (log_checkpoints)
+		elog(LOG, "checkpoint: %d buffers to write", num_to_write);
+	/* Initialize some debug code for watching progress of the checkpoint. */
+	start = GetCurrentTimestamp();
+	/* Print out message every 30 seconds */
+	targetTime = start + 30000000;
+	cur_num_written = 0;
+#endif
+
 	/*
-	 * Loop over all buffers again, and write the ones (still) marked with
-	 * BM_CHECKPOINT_NEEDED.  In this loop, we start at the clock sweep point
+	 * Loop over found dirty buffers, and write the ones (still) marked with
+	 * BM_CHECKPOINT_NEEDED. In this loop, we start at the clock sweep point
 	 * since we might as well dump soon-to-be-recycled buffers first.
 	 *
 	 * Note that we don't read the buffer alloc count here --- that should be
 	 * left untouched till the next BgBufferSync() call.
 	 */
-	buf_id = StrategySyncStart(NULL, NULL);
-	num_to_scan = NBuffers;
 	num_written = 0;
-	while (num_to_scan-- > 0)
+	for (i = 0; i < num_to_write; i++)
 	{
-		volatile BufferDesc *bufHdr = &BufferDescriptors[buf_id];
-
+		volatile BufferDesc *bufHdr;
+		buf_id = ioSeqList[seqIdx].idx;
+		bufHdr = &BufferDescriptors[buf_id];
 		/*
 		 * We don't need to acquire the lock here, because we're only looking
 		 * at a single bit. It's possible that someone else writes the buffer
@@ -1293,12 +1323,37 @@ BufferSync(int flags)
 				/*
 				 * Sleep to throttle our I/O rate.
 				 */
-				CheckpointWriteDelay(flags, (double) num_written / num_to_write);
+				if (dwInfo[DWBUF_CHECKPOINTER].endFill == dwInfo[DWBUF_CHECKPOINTER].start) {
+					/*
+					 * Prefer not to sleep while there are blocks in
+					 * double-write buffer, increasing read cost.
+					 */
+					CheckpointWriteDelay(flags, (double) num_written / num_to_write);
+				}
 			}
 		}
 
-		if (++buf_id >= NBuffers)
-			buf_id = 0;
+#ifdef DBLWRITE_DEBUG
+		if (log_checkpoints &&
+			((now = GetCurrentTimestamp()) >= targetTime ||
+			 i == num_to_write - 1))
+		{
+			/* Print out a checkpoint progress message */
+			int			progress = num_written * 100 / num_to_write;
+			int			target = (now - start) * 100 / (int) (1000000 * CheckPointTimeout * CheckPointCompletionTarget);
+
+			elog(LOG, "%d written/s, %d%%, target %d%%",
+				 (num_written - cur_num_written) / 30,
+				 progress, target);
+			cur_num_written = num_written;
+			targetTime = now + 30000000;
+		}
+#endif
+
+		if (++seqIdx >= num_to_write)
+		{
+			seqIdx = 0;
+		}
 	}
 
 	/*
@@ -1655,7 +1710,7 @@ SyncOneBuffer(int buf_id, bool skip_recently_used)
 	PinBuffer_Locked(bufHdr);
 	LWLockAcquire(bufHdr->content_lock, LW_SHARED);
 
-	FlushBuffer(bufHdr, NULL);
+	FlushBuffer(bufHdr, NULL, true);
 
 	LWLockRelease(bufHdr->content_lock);
 	UnpinBuffer(bufHdr, true);
@@ -1780,6 +1835,16 @@ CheckPointBuffers(int flags)
 	TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags);
 	CheckpointStats.ckpt_write_t = GetCurrentTimestamp();
 	BufferSync(flags);
+	/*
+	 * Before completing the checkpoint, must be sure that what's
+	 * currently in the double-write buffers is flushed.
+	 */
+	FlushDoubleWriteBuffer(DWBUF_NON_CHECKPOINTER);
+	FlushDoubleWriteBuffer(DWBUF_CHECKPOINTER);
+	elog(LOG, "(%d, %d, %d), (%d, %d, %d), waits %d",
+		 dwInfo[0].start, dwInfo[0].endFill, dwInfo[0].readHits,
+		 dwInfo[1].start, dwInfo[1].endFill, dwInfo[1].readHits,
+		 dwInfo[0].fullWaits);
 	CheckpointStats.ckpt_sync_t = GetCurrentTimestamp();
 	TRACE_POSTGRESQL_BUFFER_CHECKPOINT_SYNC_START();
 	smgrsync();
@@ -1869,7 +1934,7 @@ BufferGetTag(Buffer buffer, RelFileNode *rnode, ForkNumber *forknum,
  * if any.
  */
 static void
-FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln)
+FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln, bool needsDoubleWrite)
 {
 	XLogRecPtr	recptr;
 	ErrorContextCallback errcontext;
@@ -1924,7 +1989,7 @@ FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln)
 			  buf->tag.forkNum,
 			  buf->tag.blockNum,
 			  (char *) BufHdrGetBlock(buf),
-			  false);
+			  false, needsDoubleWrite);
 
 	pgBufferUsage.shared_blks_written++;
 
@@ -2141,10 +2206,14 @@ PrintPinnedBufs(void)
  *		used in any performance-critical code paths, so it's not worth
  *		adding additional overhead to normal paths to make it go faster;
  *		but see also DropRelFileNodeBuffers.
+ *
+ *		If needsDoubleWrite is true, then the changed buffers should use
+ *		double-writing, if that option is enabled.	Currently, relation
+ *		changes that were not WAL-logged do not need double writes.
  * --------------------------------------------------------------------
  */
 void
-FlushRelationBuffers(Relation rel)
+FlushRelationBuffers(Relation rel, bool needsDoubleWrite)
 {
 	int			i;
 	volatile BufferDesc *bufHdr;
@@ -2172,7 +2241,7 @@ FlushRelationBuffers(Relation rel)
 						  bufHdr->tag.forkNum,
 						  bufHdr->tag.blockNum,
 						  (char *) LocalBufHdrGetBlock(bufHdr),
-						  false);
+						  false, false);
 
 				bufHdr->flags &= ~(BM_DIRTY | BM_JUST_DIRTIED);
 
@@ -2196,7 +2265,7 @@ FlushRelationBuffers(Relation rel)
 		{
 			PinBuffer_Locked(bufHdr);
 			LWLockAcquire(bufHdr->content_lock, LW_SHARED);
-			FlushBuffer(bufHdr, rel->rd_smgr);
+			FlushBuffer(bufHdr, rel->rd_smgr, needsDoubleWrite && doubleWrites);
 			LWLockRelease(bufHdr->content_lock);
 			UnpinBuffer(bufHdr, true);
 		}
@@ -2238,7 +2307,7 @@ FlushDatabaseBuffers(Oid dbid)
 		{
 			PinBuffer_Locked(bufHdr);
 			LWLockAcquire(bufHdr->content_lock, LW_SHARED);
-			FlushBuffer(bufHdr, NULL);
+			FlushBuffer(bufHdr, NULL, false);
 			LWLockRelease(bufHdr->content_lock);
 			UnpinBuffer(bufHdr, true);
 		}
diff --git a/src/backend/storage/buffer/ioseq.c b/src/backend/storage/buffer/ioseq.c
new file mode 100644
index 0000000..0812ff6
--- /dev/null
+++ b/src/backend/storage/buffer/ioseq.c
@@ -0,0 +1,250 @@
+/*-------------------------------------------------------------------------
+ *
+ * ioseq.c
+ *
+ *	  Routines for sorting dirty buffers to form sequential runs.
+ *
+ * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+#include "miscadmin.h"
+#include "access/xlog.h"
+#include "storage/ioseq.h"
+
+/* GUC's sort_dirty_buffers */
+bool sortDirtyBuffers = true;
+
+/* Array of buffer indices/tags in shared memory */
+BufferIdxTag *ioSeqList;
+
+/*
+ *----------------------------------------------------------------------
+ * CompareTags --
+ *      
+ *      Comparison function that lets us group same-file buffers
+ *      together and sort on block numbers within the file. It requires
+ *      BufferTag to have no padding bytes and blockNum to be the last
+ *      field in BufferTag. See IOSort_Sort for the corresponding asserts.
+ *----------------------------------------------------------------------
+ */
+static int
+CompareTags(const void *e1, const void *e2)
+{
+	const BufferTag *t1 = &((const BufferIdxTag *)e1)->tag;
+	const BufferTag *t2 = &((const BufferIdxTag *)e2)->tag;
+	int retval = memcmp(t1, t2, offsetof(BufferTag, blockNum));
+	if (retval == 0) {
+		retval = t1->blockNum - t2->blockNum;
+	}
+	return retval;
+}
+
+/*
+ *----------------------------------------------------------------------
+ * CompareClocks --
+ *      
+ *      Comparison function that puts soon-to-be-evicted buffers earlier
+ *      in the buffer. It preserves sequential runs (they have the
+ *      same clock value) and orders by block number within each run.
+ *----------------------------------------------------------------------
+ */
+static int
+CompareClocks(const void *v1, const void *v2)
+{
+	const BufferIdxTag *e1 = (const BufferIdxTag *)v1;
+	const BufferIdxTag *e2 = (const BufferIdxTag *)v2;
+	int retval = e1->clock - e2->clock;
+	if (retval == 0) {
+		retval = e1->tag.blockNum - e2->tag.blockNum;
+	}
+	return retval;
+}
+
+/*
+ *----------------------------------------------------------------------
+ * SeqEntries --
+ *      
+ *      Returns true iff e2 belongs to the same file as e1 and its block
+ *      immediately follows e1.
+ *----------------------------------------------------------------------
+ */
+static inline bool
+SeqEntries(const BufferIdxTag *e1, const BufferIdxTag *e2)
+{
+	return memcmp(&e1->tag, &e2->tag, offsetof(BufferTag, blockNum)) == 0 &&
+		   e1->tag.blockNum + 1 == e2->tag.blockNum;
+}
+
+/*
+ *----------------------------------------------------------------------
+ * ClockVal --
+ *      
+ *      Returns the eviction priority of a given buffer index. Buffer
+ *      that will be evicted next has ClockVal == 0.
+ *----------------------------------------------------------------------
+ */
+static inline int
+ClockVal(int idx, int startIdx)
+{
+	idx -= startIdx;
+	return (idx >= 0) ? idx : idx + NBuffers;
+}
+
+/*
+ *----------------------------------------------------------------------
+ * IOSeqVerify --
+ *      
+ *      Check that the buffer list is properly sorted: if two buffers
+ *      belong to the same file and are sequential, they better be
+ *      next to each other.
+ *----------------------------------------------------------------------
+ */
+static inline void
+IOSeqVerify(BufferIdxTag *entries, unsigned numEntries)
+{
+#if 0
+	static unsigned count;
+	if (count++ % 10 == 0) {
+		unsigned i;
+		for (i = 0; i < numEntries; i++) {
+			unsigned j;
+			for (j = 0; j < numEntries; j++) {
+				Assert(!SeqEntries(&entries[i], &entries[j]) || j == i + 1);
+			}
+		}
+	}
+#endif
+}
+
+/*
+ *----------------------------------------------------------------------
+ * IOSeq_ShmemSize --
+ *
+ *      Compute the size of shared memory for the list of buffer
+ *      indices/tags.
+ *----------------------------------------------------------------------
+ */
+Size
+IOSeq_ShmemSize(void)
+{
+	return mul_size(NBuffers, sizeof(BufferIdxTag));
+}
+
+/*
+ *----------------------------------------------------------------------
+ * IOSeq_Init --
+ *
+ *      Initialize ioSeqList to point to the list of buffers'
+ *      indices/tags in shared memory. This is called once during
+ *      shared-memory initialization (either in the postmaster, or in
+ *      a standalone backend).
+ *----------------------------------------------------------------------
+ */
+void
+IOSeq_Init(void)
+{
+	bool found;
+	ioSeqList =	(BufferIdxTag *)ShmemInitStruct("Sorted Buffers",
+												IOSeq_ShmemSize(),
+												&found);
+}
+
+extern bool doubleWrites;
+
+/*
+ *----------------------------------------------------------------------
+ * IOSeq_Sort --
+ *      
+ *      Sort the passed-in list of buffers to group same-file
+ *      sequential buffers together but try to preserve the clock
+ *      order, to some extent.
+ *----------------------------------------------------------------------
+ */
+void
+IOSeq_Sort(BufferIdxTag *entries, unsigned numEntries)
+{
+	unsigned i = 0;
+	int firstVictim;
+	/*
+	 * Sort entries to put same-file blocks together. A couple of
+	 * asserts on the structure layout:
+	 * 1. BufferTag has no pad bytes -> can use memcmp in CompareTags.
+	 * 2. blockNum is the last field of BufferTag -> can use offsetof.
+	 */
+	Assert(sizeof(BufferTag) ==
+		   3 * sizeof(Oid) + sizeof(ForkNumber) + sizeof(BlockNumber) &&
+		   sizeof(BufferTag) ==
+		   offsetof(BufferTag, blockNum) + sizeof(BlockNumber));
+	qsort(entries, numEntries, sizeof(BufferIdxTag), CompareTags);
+
+	/* Assign each sequential run one clock value (min of its members) */
+	firstVictim = StrategySyncStart(NULL, NULL);
+	while (i < numEntries) {
+		int minClock = ClockVal(entries[i].idx, firstVictim);
+		unsigned j = i + 1;
+		for (; j < numEntries && SeqEntries(&entries[j-1], &entries[j]); j++) {
+			int clock = ClockVal(entries[j].idx, firstVictim);
+			if (minClock > clock) {
+				minClock = clock;
+			}
+		}
+		/* This loop also advances i past the sequential run found, so
+		 * execute it, even if we're not going to do the second qsort. */
+		for (; i < j; i++) {
+			entries[i].clock = minClock;
+		}
+	}
+
+	/*
+	 * In the case of double writes with buffered IO, don't do the
+	 * second sort, so batches are more likely to be to just a single
+	 * file, and so we only have to do one fdatasync for each batch.
+	 */
+	if (!doubleWrites) {
+		/* Re-sort runs based on the clock */
+		qsort(entries, numEntries, sizeof(BufferIdxTag), CompareClocks);
+	}
+	IOSeqVerify(entries, numEntries);
+}
+
+/*
+ *----------------------------------------------------------------------
+ * IOSeq_ClockStart --
+ *      
+ *      Finds the first dirty buffer in the clock order and returns
+ *      its position in the entries array. NB: Assumes that the
+ *      entries array is in the original order (increasing buf ids),
+ *      rather than sorted via IOSeq_Sort.
+ *----------------------------------------------------------------------
+ */
+unsigned
+IOSeq_ClockStart(BufferIdxTag *entries, unsigned numEntries)
+{
+	int val = StrategySyncStart(NULL, NULL); /* not necessarily dirty */
+	unsigned left = 0, mid, right = numEntries;
+
+	/* Binary-search to find the nearest entry with id >= val or numEntries. */
+	while (left != right) {
+		mid = (left + right) / 2;
+		if (entries[mid].idx < val) {
+			left = mid + 1;
+		} else {
+			right = mid;
+		}
+	}
+#ifdef USE_ASSERT_CHECKING
+	/* Assert the entries are sorted by idx. Also verify the binary-search. */
+	for (mid = numEntries; mid != 0 && entries[mid - 1].idx >= val; mid--) {
+		Assert(mid == numEntries || entries[mid].idx > entries[mid - 1].idx);
+	}
+	Assert(mid == left);
+#endif
+	/* If nextVictim is after the last dirty buffer, first dirty is at 0. */
+	if (left == numEntries) {
+		left = 0;
+	}
+	return left;
+}
diff --git a/src/backend/storage/buffer/localbuf.c b/src/backend/storage/buffer/localbuf.c
index 096d36a..ac7891f 100644
--- a/src/backend/storage/buffer/localbuf.c
+++ b/src/backend/storage/buffer/localbuf.c
@@ -205,7 +205,7 @@ LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum,
 				  bufHdr->tag.forkNum,
 				  bufHdr->tag.blockNum,
 				  (char *) LocalBufHdrGetBlock(bufHdr),
-				  false);
+				  false, false);
 
 		/* Mark not-dirty now in case we error out below */
 		bufHdr->flags &= ~BM_DIRTY;
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 6bad6bf..54b864d 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -56,6 +56,7 @@
 #include "pgstat.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
+#include "storage/smgr.h"
 #include "utils/guc.h"
 #include "utils/resowner.h"
 
@@ -103,7 +104,7 @@ int			max_files_per_process = 1000;
  * Note: the value of max_files_per_process is taken into account while
  * setting this variable, and so need not be tested separately.
  */
-static int	max_safe_fds = 32;	/* default if not changed */
+static int	max_safe_fds = 70;	/* default if not changed */
 
 
 /* Debugging.... */
@@ -1332,6 +1333,233 @@ retry:
 	return returnCode;
 }
 
+struct io_iocb_common
+{
+	void	   *buf;
+	unsigned	nbytes;
+	long long	offset;
+};
+
+struct iocb
+{
+	void	   *data;
+	int			aio_fildes;
+	union
+	{
+		struct io_iocb_common c;
+	}			u;
+};
+
+/* Array of iocbs used for io_submit */
+struct iocb ioc[MAX_BATCH_BLOCKS];
+int			iocIndex = 0;
+
+/* Array of iovecs use for iocbs and for writev.  One extra entry for
+ * double-write header. */
+struct iovec iov[MAX_BATCH_BLOCKS + 1];
+int			iovIndex = 0;
+
+/*
+ * Return number of bytes described by the list of memory regions (struct
+ * iovec elements) referenced by ioc.
+ */
+static int
+iovlen(struct iocb *ioc)
+{
+	int			i,
+				len;
+	struct iovec *iov = (struct iovec *) ioc->u.c.buf;
+
+	len = 0;
+	for (i = 0; i < ioc->u.c.nbytes; i++)
+	{
+		len += iov->iov_len;
+		iov++;
+	}
+	return len;
+}
+
+/*
+ * Flush existing batched IOs, doing a write to a double-write file first
+ * if double_writes option is turned on.  When called, iov[] lists all
+ * the memory regions being written, in order, and ioc[] groups together
+ * consecutive iov elements which go to consecutive locations on disk.
+ * Each ioc represents a set of memory regions that can be written to
+ * disk in a single write using writev().
+ */
+static int
+FlushIOs(File doubleWriteFile)
+{
+	int			returnCode;
+	int			i,
+				j;
+
+	if (doubleWriteFile > 0)
+	{
+		returnCode = FileAccess(doubleWriteFile);
+		if (returnCode < 0)
+			return returnCode;
+		returnCode = lseek(VfdCache[doubleWriteFile].fd, 0, SEEK_SET);
+		if (returnCode < 0)
+		{
+			elog(LOG, "lseek error errno %d, rc %d",
+				 errno, returnCode);
+			return returnCode;
+		}
+
+		/*
+		 * Write out all the blocks sequentially in double-write file,
+		 * starting with the double-write file header describing all the
+		 * buffers.
+		 */
+		returnCode = writev(VfdCache[doubleWriteFile].fd, &iov[0], iovIndex);
+		if (returnCode < 0)
+		{
+			elog(LOG, "writev to tmp, errno %d, rc %d, iovIndex %d",
+				 errno, returnCode, iovIndex);
+			return returnCode;
+		}
+#if 0
+		returnCode = fdatasync(VfdCache[doubleWriteFile].fd);
+		if (returnCode < 0)
+		{
+			elog(LOG, "fdatasync error errno %d, rc %d", errno,
+				 returnCode);
+			return returnCode;
+		}
+#endif
+	}
+
+	for (i = 0; i < iocIndex; i++)
+	{
+		returnCode = lseek(ioc[i].aio_fildes, ioc[i].u.c.offset, SEEK_SET);
+		if (returnCode < 0)
+		{
+			elog(LOG, "lseek error errno %d, rc %d, offset %Ld",
+				 errno, returnCode, ioc[i].u.c.offset);
+			return returnCode;
+		}
+		/* Use writev to do one IO per consecutive blocks to the disk. */
+		returnCode = writev(ioc[i].aio_fildes, ioc[i].u.c.buf,
+							ioc[i].u.c.nbytes);
+		if (returnCode < 0)
+		{
+			elog(LOG, "writev error errno %d, rc %d", errno, returnCode);
+			return returnCode;
+		}
+	}
+	if (doubleWriteFile > 0)
+	{
+		/*
+		 * If we are doing double writes, must make sure that the blocks was
+		 * just wrote are forced to disk, so we can re-use double-write buffer
+		 * next time.
+		 */
+		for (i = 0; i < iocIndex; i++)
+		{
+			for (j = 0; j < i; j++)
+			{
+				if (ioc[j].aio_fildes == ioc[i].aio_fildes)
+				{
+					break;
+				}
+			}
+			if (j == i)
+			{
+				returnCode = fdatasync(ioc[i].aio_fildes);
+				if (returnCode < 0)
+				{
+					elog(LOG, "fdatasync error errno %d, rc %d", errno,
+						 returnCode);
+					return returnCode;
+				}
+			}
+		}
+	}
+	return 0;
+}
+
+/*
+ * Write out a batch of memory regions (Postgres buffers) to locations in
+ * files, as specified in writeList.  If doubleWriteFile is >= 0, then
+ * also do double writes to the specified file (so full_page_writes can
+ * be avoided).
+ */
+int
+FileBwrite(int writeLen, struct SMgrWriteList *writeList,
+		   File doubleWriteFile, char *doubleBuf)
+{
+	int			returnCode;
+	int			i;
+
+	for (i = 0; i < writeLen; i++)
+	{
+		struct SMgrWriteList *w = &(writeList[i]);
+
+		Assert(FileIsValid(w->fd));
+		DO_DB(elog(LOG, "FileBwrite: %d (%s) " INT64_FORMAT " %d %p",
+				   file, VfdCache[w->fd].fileName,
+				   (int64) VfdCache[w->fd].seekPos,
+				   w->len, w->callerBuf));
+
+		returnCode = FileAccess(w->fd);
+		if (returnCode < 0) {
+			return returnCode;
+		}
+	}
+
+	iovIndex = 0;
+	iocIndex = 0;
+
+	if (doubleWriteFile > 0)
+	{
+		/*
+		 * Write out the double-write header (which lists all the blocks) in a
+		 * 4K chunk at the beginning of the double-write file.
+		 */
+		iov[iovIndex].iov_base = doubleBuf;
+		iov[iovIndex].iov_len = DOUBLE_WRITE_HEADER_SIZE;
+		iovIndex++;
+	}
+
+	/*
+	 * Convert the list of writes (writeList) into an iovec array iov that
+	 * describes all the memory regions being written, and an ioc array for
+	 * each consecutive set of iov elements which are going to the consecutive
+	 * locations in the same file.	Each ioc element describes a set of memory
+	 * regions that can be written to the disk in a single write usinf
+	 * writev() (also known as a scatter-gather array).
+	 */
+	for (i = 0; i < writeLen; i++)
+	{
+		struct SMgrWriteList *w = &(writeList[i]);
+		struct iocb *last;
+
+		iov[iovIndex].iov_base = w->buffer;
+		iov[iovIndex].iov_len = w->len;
+		last = (iocIndex > 0) ? &(ioc[iocIndex - 1]) : NULL;
+		if (iocIndex > 0 && w->seekPos == last->u.c.offset + iovlen(last) &&
+			last->aio_fildes == VfdCache[w->fd].fd)
+		{
+			last->u.c.nbytes++;
+		}
+		else
+		{
+			ioc[iocIndex].aio_fildes = VfdCache[w->fd].fd;
+			ioc[iocIndex].u.c.buf = (void *) &iov[iovIndex];
+			ioc[iocIndex].u.c.nbytes = 1;
+			ioc[iocIndex].u.c.offset = w->seekPos;
+			iocIndex++;
+		}
+		iovIndex++;
+		VfdCache[w->fd].seekPos = FileUnknownPos;
+	}
+
+	returnCode = FlushIOs(doubleWriteFile);
+
+	return returnCode;
+}
+
 int
 FileSync(File file)
 {
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index ef1dc91..88a8819 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -37,7 +37,7 @@
 #include "storage/procsignal.h"
 #include "storage/sinvaladt.h"
 #include "storage/spin.h"
-
+#include "storage/ioseq.h"
 
 shmem_startup_hook_type shmem_startup_hook = NULL;
 
@@ -105,6 +105,8 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 		size = add_size(size, hash_estimate_size(SHMEM_INDEX_SIZE,
 												 sizeof(ShmemIndexEnt)));
 		size = add_size(size, BufferShmemSize());
+		size = add_size(size, IOSeq_ShmemSize());
+		size = add_size(size, (NON_CKPT_DW_BLOCKS + CKPT_DW_BLOCKS) * BLCKSZ + BLCKSZ);
 		size = add_size(size, LockShmemSize());
 		size = add_size(size, PredicateLockShmemSize());
 		size = add_size(size, ProcGlobalShmemSize());
@@ -194,6 +196,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 	SUBTRANSShmemInit();
 	MultiXactShmemInit();
 	InitBufferPool();
+	IOSeq_Init();
 
 	/*
 	 * Set up lock manager
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index cc41568..d38c716 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -31,7 +31,7 @@
 #include "storage/predicate.h"
 #include "storage/proc.h"
 #include "storage/spin.h"
-
+#include "storage/smgr.h"
 
 /* We use the ShmemLock spinlock to protect LWLockAssign */
 extern slock_t *ShmemLock;
@@ -186,6 +186,12 @@ NumLWLocks(void)
 	numLocks += NUM_OLDSERXID_BUFFERS;
 
 	/*
+	 * For double-write buffers:  2 writeLocks, 1 allocLock, and one lock
+	 * per batch in the double-write buffers.
+	 */
+	numLocks += 3 + 2 * MAX_BATCHES_PER_DWBUF;
+
+	/*
 	 * Add any requested by loadable modules; for backwards-compatibility
 	 * reasons, allocate at least NUM_USER_DEFINED_LWLOCKS of them even if
 	 * there are no explicit requests.
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index bfc9f06..6856856 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -745,6 +745,55 @@ mdwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 }
 
 /*
+ * Write out a list of buffers, as specified in writeList.	If
+ * doubleWriteFile is >= 0, then also do double writes to the specified
+ * file (so full_page_writes can be avoided).
+ */
+void
+mdbwrite(int writeLen, struct SMgrWriteList *writeList,
+		 File doubleWriteFile, char *doubleBuf)
+{
+	off_t		seekpos;
+	int			nbytes;
+	MdfdVec    *v;
+	File		fd;
+	int			i;
+
+	for (i = 0; i < writeLen; i++)
+	{
+		struct SMgrWriteList *w = &(writeList[i]);
+
+		/* This assert is too expensive to have on normally ... */
+#ifdef CHECK_WRITE_VS_EXTEND
+		Assert(w->blockNum < mdnblocks(w->localrel, w->forknum));
+#endif
+		v = _mdfd_getseg(w->localrel, w->forkNum, w->blockNum, false, EXTENSION_FAIL);
+		fd = v->mdfd_vfd;
+
+		seekpos = (off_t) BLCKSZ *(w->blockNum % ((BlockNumber) RELSEG_SIZE));
+
+		Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+		w->fd = fd;
+		w->seekPos = seekpos;
+		w->len = BLCKSZ;
+	}
+
+	nbytes = FileBwrite(writeLen, writeList, doubleWriteFile, doubleBuf);
+	if (nbytes < 0)
+	{
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("FileBwrite error %d", errno)));
+	}
+
+	/*
+	 * XXX register_dirty_segment() call is not needed if we only do batched
+	 * writes for the double_write option.
+	 */
+}
+
+
+/*
  *	mdnblocks() -- Get the number of blocks stored in a relation.
  *
  *		Important side effect: all active segments of the relation are opened
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 5f87543..9df4fd1 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -15,10 +15,15 @@
  *
  *-------------------------------------------------------------------------
  */
+#include <unistd.h>
+
 #include "postgres.h"
 
+#include "catalog/catalog.h"
 #include "commands/tablespace.h"
+#include "postmaster/bgwriter.h"
 #include "storage/bufmgr.h"
+#include "storage/barrier.h"
 #include "storage/ipc.h"
 #include "storage/smgr.h"
 #include "utils/hsearch.h"
@@ -53,6 +58,8 @@ typedef struct f_smgr
 										  BlockNumber blocknum, char *buffer);
 	void		(*smgr_write) (SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum, char *buffer, bool skipFsync);
+	void		(*smgr_bwrite) (int writeLen, struct SMgrWriteList *writeList,
+									  File doubleWriteFile, char *doubleBuf);
 	BlockNumber (*smgr_nblocks) (SMgrRelation reln, ForkNumber forknum);
 	void		(*smgr_truncate) (SMgrRelation reln, ForkNumber forknum,
 											  BlockNumber nblocks);
@@ -66,7 +73,7 @@ typedef struct f_smgr
 static const f_smgr smgrsw[] = {
 	/* magnetic disk */
 	{mdinit, NULL, mdclose, mdcreate, mdexists, mdunlink, mdextend,
-		mdprefetch, mdread, mdwrite, mdnblocks, mdtruncate, mdimmedsync,
+		mdprefetch, mdread, mdwrite, mdbwrite, mdnblocks, mdtruncate, mdimmedsync,
 		mdpreckpt, mdsync, mdpostckpt
 	}
 };
@@ -79,9 +86,30 @@ static const int NSmgr = lengthof(smgrsw);
  */
 static HTAB *SMgrRelationHash = NULL;
 
+/* Page checksumming. */
+static uint64 tempbuf[BLCKSZ / sizeof(uint64)];
+static bool FlushPreviousBatches(struct DWBufferInfo *dwi, int endFill);
+
+#define INVALID_CKSUM 0x1b0af034
+
 /* local function prototypes */
 static void smgrshutdown(int code, Datum arg);
 
+/*
+ * Buffer used to write the double-write header.  Useful to align buffer
+ * to page size if it is written with direct IO.
+ */
+#define CC_ALIGN(_LEN) __attribute__ ((aligned (_LEN)))
+static char doubleBuf[DOUBLE_WRITE_HEADER_SIZE] CC_ALIGN(4096);
+
+/* Files to which double writes are going */
+File		doubleWriteFile[16];
+
+/* Info on the double-write buffers */
+struct DWBufferInfo *dwInfo;
+/* Blocks of the double-write buffers */
+char *dwBlocks;
+
 
 /*
  *	smgrinit(), smgrshutdown() -- Initialize or shut down storage
@@ -329,6 +357,43 @@ smgrcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo)
 }
 
 /*
+ * Invalidate entries in the double-write buffer that are writing
+ * to the specified relation.
+ */
+static void
+InvalDoubleWriteBuffer(SMgrRelation reln, ForkNumber forknum, int dwIndex)
+{
+	struct DWBufferInfo *dwi = dwInfo + dwIndex;
+	int endFill, i;
+
+	LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+	endFill = dwi->endFill;
+	if (FlushPreviousBatches(dwi, endFill))
+	{
+		/*
+		 * We've flushed everything up to endFill, so nothing to invalidate.
+		 */
+		LWLockRelease(dwi->writeLock);
+	}
+	else
+	{
+		Assert(dwi->start <= endFill);
+		for (i = dwi->start; i < endFill; i++)
+		{
+			struct SMgrWriteList *wr = &(dwi->writeList[i & dwi->mask]);
+			if (memcmp(&(wr->smgr_rnode), &(reln->smgr_rnode),
+					   sizeof(RelFileNodeBackend)) == 0 &&
+				wr->forkNum == forknum)
+			{
+				elog(LOG, "Invalidating entry %d in dwbuf %d", i, dwIndex);
+				wr->forkNum = InvalidForkNumber;
+			}
+		}
+		LWLockRelease(dwi->writeLock);
+	}
+}
+
+/*
  *	smgrdounlink() -- Immediately unlink a relation.
  *
  *		The specified fork of the relation is removed from the store.  This
@@ -353,6 +418,15 @@ smgrdounlink(SMgrRelation reln, ForkNumber forknum, bool isRedo)
 	 */
 	DropRelFileNodeBuffers(rnode, forknum, 0);
 
+	if (doubleWrites) {
+		/*
+		 * Can invalidate one at a time, since relation is already
+		 * cleared from the buffer cache.
+		 */
+		InvalDoubleWriteBuffer(reln, forknum, DWBUF_NON_CHECKPOINTER);
+		InvalDoubleWriteBuffer(reln, forknum, DWBUF_CHECKPOINTER);
+	}
+
 	/*
 	 * It'd be nice to tell the stats collector to forget it immediately, too.
 	 * But we can't because we don't know the OID (and in cases involving
@@ -381,6 +455,61 @@ smgrdounlink(SMgrRelation reln, ForkNumber forknum, bool isRedo)
 }
 
 /*
+ * The initial value when computing the checksum for a data page.
+ */
+static inline uint64
+ChecksumInit(SMgrRelation reln, ForkNumber f, BlockNumber b)
+{
+	return b + f;
+}
+
+/*
+ * Compute a checksum of buffer (with length len), using initial value
+ * cksum.  We use a relatively simple checksum calculation to avoid
+ * overhead, but could replace with some kind of CRC calculation.
+ */
+static inline uint32
+ComputeChecksum(uint64 *buffer, uint32 len, uint64 cksum)
+{
+	int			i;
+
+	for (i = 0; i < len / sizeof(uint64); i += 4)
+	{
+		cksum += (cksum << 5) + *buffer;
+		cksum += (cksum << 5) + *(buffer + 1);
+		cksum += (cksum << 5) + *(buffer + 2);
+		cksum += (cksum << 5) + *(buffer + 3);
+		buffer += 4;
+	}
+	cksum = (cksum & 0xFFFFFFFF) + (cksum >> 32);
+	return cksum;
+}
+
+/*
+ * Copy buffer to dst and compute the checksum during the copy (so that
+ * the checksum is correct for the final contents of dst).
+ */
+static inline uint32
+CopyAndComputeChecksum(uint64 *dst, volatile uint64 *buffer,
+					   uint32 len, uint64 cksum)
+{
+	int			i;
+
+	for (i = 0; i < len / sizeof(uint64); i += 4)
+	{
+		cksum += (cksum << 5) + (*dst = *buffer);
+		cksum += (cksum << 5) + (*(dst + 1) = *(buffer + 1));
+		cksum += (cksum << 5) + (*(dst + 2) = *(buffer + 2));
+		cksum += (cksum << 5) + (*(dst + 3) = *(buffer + 3));
+		dst += 4;
+		buffer += 4;
+	}
+	cksum = (cksum & 0xFFFFFFFF) + (cksum >> 32);
+	return cksum;
+}
+
+
+/*
  *	smgrextend() -- Add a new block to a file.
  *
  *		The semantics are nearly the same as smgrwrite(): write at the
@@ -393,8 +522,30 @@ void
 smgrextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		   char *buffer, bool skipFsync)
 {
+	PageHeader	p;
+
+	Assert(PageGetPageLayoutVersion(((PageHeader) buffer)) == PG_PAGE_LAYOUT_VERSION ||
+		   PageIsNew(buffer));
+	if (page_checksum)
+	{
+		p = (PageHeader) tempbuf;
+		((PageHeader) buffer)->cksum = 0;
+
+		/*
+		 * We copy and compute the checksum, and then write out the data from
+		 * the copy, so that we avoid any problem with hint bits changing
+		 * after we compute the checksum.
+		 */
+		p->cksum = CopyAndComputeChecksum(tempbuf, (uint64 *) buffer, BLCKSZ,
+									  ChecksumInit(reln, forknum, blocknum));
+	}
+	else
+	{
+		p = (PageHeader) buffer;
+		p->cksum = INVALID_CKSUM;
+	}
 	(*(smgrsw[reln->smgr_which].smgr_extend)) (reln, forknum, blocknum,
-											   buffer, skipFsync);
+											   (char *) p, skipFsync);
 }
 
 /*
@@ -407,6 +558,62 @@ smgrprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 }
 
 /*
+ * Look for specified reln/fork/block in the double-write buffer.  Return
+ * TRUE and the index in the buffer if found.
+ */
+static bool
+CheckDoubleWriteBuffer(int dwIndex, SMgrRelation reln, ForkNumber forknum,
+					   BlockNumber blocknum, int *match)
+{
+	struct DWBufferInfo *dwi = dwInfo + dwIndex;
+	int i;
+	int start, endFill;
+
+	/*
+	 * Check without lock first.  If we see what we think is a match, we
+	 * will check again with appropriate lock.
+	 */
+	pg_memory_barrier();
+	start = dwi->start;
+	endFill = dwi->endFill;
+	for (i = start; i < endFill; i++)
+	{
+		struct SMgrWriteList *p = &(dwi->writeList[i & dwi->mask]);
+
+		if (p->blockNum == blocknum &&
+			memcmp(&(p->smgr_rnode), &(reln->smgr_rnode),
+				   sizeof(RelFileNodeBackend)) == 0 &&
+			p->forkNum == forknum) {
+			break;
+		}
+	}
+	if (i == endFill) {
+		return FALSE;
+	}
+	
+	LWLockAcquire(dwi->writeLock, LW_SHARED);
+	start = dwi->start;
+	endFill = dwi->endFill;
+	Assert(start <= endFill);
+	for (i = start; i < endFill; i++)
+	{
+		struct SMgrWriteList *p = &(dwi->writeList[i & dwi->mask]);
+
+		if (p->blockNum == blocknum &&
+			memcmp(&(p->smgr_rnode), &(reln->smgr_rnode),
+				   sizeof(RelFileNodeBackend)) == 0 &&
+			p->forkNum == forknum) {
+			if (0) elog(LOG, "Found block in the double-write buffer %d", dwIndex);
+			dwi->readHits++;
+			*match = i;
+			return TRUE;
+		}
+	}
+	LWLockRelease(dwi->writeLock);
+	return FALSE;
+}
+
+/*
  *	smgrread() -- read a particular block from a relation into the supplied
  *				  buffer.
  *
@@ -416,9 +623,308 @@ smgrprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
  */
 void
 smgrread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
-		 char *buffer)
+		 char *buffer, bool *cksumMismatch)
 {
+	PageHeader	p = (PageHeader) buffer;
+
+	if (doubleWrites)
+	{
+		int match;
+		struct DWBufferInfo *dwi;
+
+		/*
+		 * Must look for the block first in the double-write buffers.
+		 * Can check one at a time, since no one else can be reading the
+		 * buffer into memory.
+		 */
+		dwi = dwInfo + DWBUF_CHECKPOINTER;
+		if (CheckDoubleWriteBuffer(DWBUF_CHECKPOINTER, reln, forknum,
+								   blocknum, &match))
+		{
+			memcpy(buffer, dwi->writeList[match & dwi->mask].buffer, BLCKSZ);
+			LWLockRelease(dwi->writeLock);
+			return;
+		}
+		dwi = dwInfo + DWBUF_NON_CHECKPOINTER;
+		if (CheckDoubleWriteBuffer(DWBUF_NON_CHECKPOINTER, reln, forknum,
+								   blocknum, &match))
+		{
+			memcpy(buffer, dwi->writeList[match & dwi->mask].buffer, BLCKSZ);
+			LWLockRelease(dwi->writeLock);
+				return;
+		}
+	}
+
 	(*(smgrsw[reln->smgr_which].smgr_read)) (reln, forknum, blocknum, buffer);
+	Assert(PageIsNew(p) || PageGetPageLayoutVersion(p) == PG_PAGE_LAYOUT_VERSION);
+	if (page_checksum && p->cksum != INVALID_CKSUM)
+	{
+		const uint32 diskCksum = p->cksum;
+		uint32		cksum;
+
+		p->cksum = 0;
+		cksum = ComputeChecksum((uint64 *) buffer, BLCKSZ,
+								ChecksumInit(reln, forknum, blocknum));
+		if (cksum != diskCksum)
+		{
+			if (cksumMismatch != NULL)
+			{
+				*cksumMismatch = TRUE;
+				return;
+			}
+			ereport(PANIC, (0, errmsg("checksum mismatch: disk has %#x, should be %#x\n"
+				  "filename %s, BlockNum %u, block specifier %d/%d/%d/%d/%u",
+									  diskCksum, (uint32) cksum,
+									  relpath(reln->smgr_rnode, forknum),
+									  blocknum,
+									  reln->smgr_rnode.node.spcNode,
+									  reln->smgr_rnode.node.dbNode,
+									  reln->smgr_rnode.node.relNode,
+									  forknum, blocknum)));
+		}
+	}
+}
+
+/*
+ * If double-write file is not yet open in this process, open up the
+ * double-write file.
+ */
+static void
+CheckDoubleWriteFile(int index)
+{
+	if (doubleWriteFile[index] == 0)
+	{
+		char	   *name = DoubleWriteFileName(index);
+
+		doubleWriteFile[index] = PathNameOpenFile(name,
+												  O_RDWR | O_CREAT | O_DIRECT,
+												  S_IRUSR | S_IWUSR);
+		if (doubleWriteFile[index] < 0)
+			elog(PANIC, "Couldn't open double-write file %s", name);
+		pfree(name);
+	}
+}
+
+/*
+ * Flush blocks start to end-1 in the double-write buffer.
+ */
+static void
+FlushDoubleWriteBatch(int dwIndex, int start, int end)
+{
+	struct DWBufferInfo * dwi = dwInfo + dwIndex;
+	int fileIndex = (start & dwi->mask) / batched_buffer_writes +
+		MAX_BATCHES_PER_DWBUF * dwIndex;
+
+	CheckDoubleWriteFile(fileIndex);
+	/*
+	 * The partial write list we need is contiguous because
+	 * batched_buffer_writes always evenly divides MAX_DW_BLOCKS.
+	 */
+	Assert(start < end && end - start <= batched_buffer_writes);
+	//elog(LOG, "Flushing %d, %d", start, end);
+	smgrbwrite(end - start, &(dwi->writeList[start & dwi->mask]),
+			   doubleWriteFile[fileIndex]);
+}
+
+/*
+ * Flush the batch ending at endFlush if no other process is doing it.
+ * If doWait is TRUE, then wait until batch is definitely flushed.
+ * endFlush should be the end of a batch (divisible by
+ * batched_buffer_writes)
+ */
+static void
+TryFlushBatch(int dwIndex, int endFlush, bool doWait)
+{
+	struct DWBufferInfo *dwi = dwInfo + dwIndex;
+	LWLockId batchLock;
+	int start;
+
+	pg_memory_barrier();
+	if (dwi->start >= endFlush)
+	{
+		return;
+	}
+	batchLock = dwi->batchLocks[endFlush & dwi->mask];
+	/* XXX Race - might end up waiting for a later flush? */
+	if (doWait)
+	{
+		/*
+		 * Getting the batch lock will wait for batch to be flushed or
+		 * allow us to do it.
+		 */
+		LWLockAcquire(batchLock, LW_EXCLUSIVE);
+	}
+	else
+	{
+		if (!LWLockConditionalAcquire(batchLock, LW_EXCLUSIVE))
+		{
+			/* Someone else is flushing -- let them do it... */
+			return;
+		}
+	}
+
+	if (dwi->start >= endFlush || dwi->flushed[endFlush & dwi->mask])
+	{
+		/* Some other process already flushed the batch */
+		LWLockRelease(batchLock);
+		return;
+	}
+
+	/* Find beginning of batch */
+	start = endFlush - batched_buffer_writes;
+	if (start < dwi->start)
+	{
+		start = dwi->start;
+	}
+	FlushDoubleWriteBatch(dwIndex, start, endFlush);
+	LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+	if (start == dwi->start)
+	{
+		/*
+		 * We flushed the oldest block, so let's advance dwi->start.
+		 */
+		dwi->start = endFlush;
+		dwi->flushed[endFlush & dwi->mask] = 0;
+		/*
+		 * Keep advancing dwi->start if there were later blocks
+		 * flushed.
+		 */
+		while (dwi->endFill >= endFlush + batched_buffer_writes &&
+			   dwi->flushed[(endFlush + batched_buffer_writes) & dwi->mask])
+		{
+			endFlush += batched_buffer_writes;
+			dwi->start = endFlush;
+			dwi->flushed[endFlush & dwi->mask] = 0;
+		}
+	}
+	else
+	{
+		/* Not the oldest block, so just mark that we flushed it. */
+		dwi->flushed[endFlush & dwi->mask] = 1;
+	}
+	LWLockRelease(dwi->writeLock);
+	LWLockRelease(batchLock);
+}	
+
+
+/* Round down a to the boundary of a batch */
+#define ROUND_DOWN_BATCH(a)  ((a) - ((a) % batched_buffer_writes))
+#define ROUND_UP_BATCH(a)    ((a) + batched_buffer_writes - ((a) % batched_buffer_writes))
+
+/*
+ * Flush all batches previous to endFill.  Requires dwi->writeLock is
+ * held.  Return TRUE, if, while waiting, we actually flushed the batch
+ * containing endFill itself.
+ */
+static bool
+FlushPreviousBatches(struct DWBufferInfo *dwi, int endFill)
+{
+	if (endFill == dwi->start) {
+		/* Nothing to flush */
+		return TRUE;
+	}
+	if (dwi->start >= ROUND_DOWN_BATCH(endFill)) {
+		/* No previous batches to flush */
+		return FALSE;
+	}
+	
+	/*
+	 * Other processes should already be flushing up to
+	 * ROUND_DOWN_BATCH(endFill)
+	 */
+	LWLockRelease(dwi->writeLock);
+	
+	while (dwi->start < ROUND_DOWN_BATCH(endFill)) {
+		//elog(LOG, "Waiting for flushes in FlushPreviousBatches");
+		TryFlushBatch(dwi - dwInfo, ROUND_UP_BATCH(dwi->start), TRUE);
+	}
+	LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+	if (ROUND_DOWN_BATCH(dwi->endFill) >= endFill) {
+		/* dwi->endFill has now advanced past endFill's batch, so we can
+		 * wait for someone to flush past endFill */
+		if (dwi->start < endFill) {
+			LWLockRelease(dwi->writeLock);
+			while (dwi->start < endFill) {
+				//elog(LOG, "Waiting for last flush in FlushPreviousBatches");
+				TryFlushBatch(dwi - dwInfo, ROUND_UP_BATCH(dwi->start), TRUE);
+			}
+			LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+		}
+		return TRUE;
+	}
+	return FALSE;
+}
+
+/*
+ * Check if there is a duplicate for the specified reln/fork/block
+ * already in the double-write buffer.  If so, handle it (wait for it to
+ * be flushed or copy over previous contents with current contents).
+ */
+static void
+CheckDuplicate(int dwIndex, SMgrRelation reln, ForkNumber forknum,
+			   BlockNumber blocknum, char *buffer)
+{
+	int match;
+
+	if (CheckDoubleWriteBuffer(dwIndex, reln, forknum, blocknum, &match))
+	{
+		struct DWBufferInfo *dwi = dwInfo + dwIndex;
+
+		elog(LOG, "Duplicate block in dwbuf [%d] %d (%d %d)",
+			 dwIndex, match, dwi->start, dwi->endFill);
+		if (match < ROUND_DOWN_BATCH(dwi->endFill))
+		{
+			/* Duplicate is in previous batch, so wait for that to be flushed. */
+			LWLockRelease(dwi->writeLock);
+			while (dwi->start <= match)
+			{
+				elog(LOG, "Waiting for flush because of duplicate");
+				TryFlushBatch(dwIndex, ROUND_UP_BATCH(dwi->start), TRUE);
+			}
+		}
+		else
+		{
+			/*
+			 * Duplicate is in this batch, just copy over previous
+			 * contents, so both writes will do the same thing (instead
+			 * of removing previous write, which will be harder to do).
+			 */
+			memcpy(buffer, dwi->writeList[match & dwi->mask].buffer, BLCKSZ);
+			LWLockRelease(dwi->writeLock);
+		}
+	}
+}
+
+/*
+ * Flush double-write buffer up to its current fill length
+ */
+void
+FlushDoubleWriteBuffer(int dwIndex)
+{
+	int endFill;
+	struct DWBufferInfo *dwi = dwInfo + dwIndex;
+
+	LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+	endFill = dwi->endFill;
+	if (FlushPreviousBatches(dwi, endFill))
+	{
+		LWLockRelease(dwi->writeLock);
+		return;
+	}
+
+	/* We must flush the last partial batch */
+	if (0) elog(LOG, "Flushing partial batch [%d] %d, %d %d %d %d in FlushDoubleWriteBuffer",
+				dwIndex, endFill - dwi->start,
+				dwi->start, endFill, dwi->endFill, dwi->endAlloc);
+	Assert(endFill < dwi->start + batched_buffer_writes);
+	/*
+	 * Keep write lock while flushing partial batch, so no one else
+	 * can add to the batch and complete it.
+	 */
+	FlushDoubleWriteBatch(dwIndex, dwi->start, endFill);
+	dwi->flushed[ROUND_UP_BATCH(dwi->start) & dwi->mask] = 0;
+	dwi->start = endFill;
+	LWLockRelease(dwi->writeLock);
 }
 
 /*
@@ -438,10 +944,203 @@ smgrread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
  */
 void
 smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
-		  char *buffer, bool skipFsync)
+		  char *buffer, bool skipFsync, bool dw)
+{
+	PageHeader	p;
+	int myBlock = 0;
+	int dwIndex = 0;
+	struct DWBufferInfo *dwi = NULL;
+
+	dw = dw && doubleWrites;
+	if (dw)
+	{
+		dwIndex = AmCheckpointer() ? DWBUF_CHECKPOINTER : DWBUF_NON_CHECKPOINTER;
+		dwi = &dwInfo[dwIndex];
+
+		/*
+		 * Since no other process can be writing out this buffer, we
+		 * just need to check the double-write buffers for duplicates
+		 * once upfront.
+		 */
+		CheckDuplicate(DWBUF_CHECKPOINTER, reln, forknum, blocknum, buffer);
+		CheckDuplicate(DWBUF_NON_CHECKPOINTER, reln, forknum, blocknum, buffer);
+
+		if (dwIndex == DWBUF_CHECKPOINTER)
+		{
+			/* No locking needed */
+			myBlock = dwi->endAlloc++;
+		}
+		else
+		{
+			/*
+			 * Get the next DW block that we can copy into.
+			 * allocLock ensures there is no starvation, and we
+			 * will get a block once one is available.
+			 */
+			LWLockAcquire(dwi->allocLock, LW_EXCLUSIVE);
+			if (dwi->endAlloc - dwi->start >= dwi->length)
+			{
+				//elog(LOG, "Waiting for too many blocks in double-write buffer");
+				dwi->fullWaits++;
+				TryFlushBatch(DWBUF_NON_CHECKPOINTER, ROUND_UP_BATCH(dwi->start),
+							  TRUE);
+				pg_memory_barrier();
+				Assert(dwi->endAlloc - dwi->start < dwi->length);
+			}
+			/*
+			 * Now that we've allocated an entry, get the write lock to
+			 * copy in the buffer and add the new entry.
+			 */
+			LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+			Assert(dwi->endAlloc >= dwi->endFill);
+			myBlock = dwi->endAlloc++;
+		}
+		p = (PageHeader)(dwi->bufferStart + BLCKSZ * (myBlock & dwi->mask));
+	}
+	else if (page_checksum)
+	{
+		/* Checksums to tempbuf, since no double-writing. */
+		p = (PageHeader) tempbuf;
+	}
+	else
+	{
+		/* No checksums or double-writing. */
+		p = (PageHeader) buffer;
+	}
+		
+	if (page_checksum)
+	{
+		((PageHeader) buffer)->cksum = 0;
+
+		/*
+		 * We copy and compute the checksum, and then write out the data from
+		 * the copy, so that we avoid any problem with hint bits changing
+		 * after we compute the checksum.
+		 */
+		p->cksum = CopyAndComputeChecksum((uint64 *)p, (uint64 *) buffer,
+										  BLCKSZ,
+										  ChecksumInit(reln, forknum, blocknum));
+	}
+	else
+	{
+		if (dw)
+		{
+			/* No checksum, just copy to double-write buffer */
+			memcpy(p, buffer, BLCKSZ);
+		}
+		p->cksum = INVALID_CKSUM;
+	}
+
+	Assert(PageGetPageLayoutVersion(p) == PG_PAGE_LAYOUT_VERSION);
+
+	if (dw)
+	{
+		int i;
+		struct SMgrWriteList *wr;
+		int endFlush;
+		bool tryFlush;
+
+		if (dwIndex == DWBUF_CHECKPOINTER)
+		{
+			/* For the checkpointer, we hadn't gotten writeLock yet. */
+			LWLockAcquire(dwi->writeLock, LW_EXCLUSIVE);
+		}
+		Assert(dwi->start <= dwi->endFill);
+
+		/* Add info on the block we want to write to the writeList */
+		i = dwi->endFill++;
+		wr = &(dwi->writeList[i & dwi->mask]);
+		wr->smgr_rnode = reln->smgr_rnode;
+		wr->forkNum = forknum;
+		wr->blockNum = blocknum;
+		wr->buffer = (char *)p;
+		wr->slot = myBlock & dwi->mask;
+
+		endFlush = dwi->endFill;
+		tryFlush = (endFlush > dwi->start && (endFlush % batched_buffer_writes) == 0);
+		LWLockRelease(dwi->writeLock);
+		if (dwIndex == DWBUF_NON_CHECKPOINTER)
+		{
+			LWLockRelease(dwi->allocLock);
+		}
+		if (tryFlush)
+		{
+			/* We filled out a batch, so let's flush it */
+			TryFlushBatch(dwIndex, endFlush, FALSE);
+		}
+	}
+	else
+	{
+		(*(smgrsw[reln->smgr_which].smgr_write)) (reln, forknum, blocknum,
+												  (char *) p, skipFsync);
+	}
+}
+
+/*
+ * Write out a list of buffers, as specified in writeList.	If
+ * doubleWriteFile is >= 0, then also do double writes to the specified
+ * file (so full_page_writes can be avoided).
+ */
+void
+smgrbwrite(int writeLen, struct SMgrWriteList *writeList,
+		   File doubleWriteFile)
 {
-	(*(smgrsw[reln->smgr_which].smgr_write)) (reln, forknum, blocknum,
-											  buffer, skipFsync);
+	PageHeader	p = NULL;
+	int			i;
+
+	/*
+	 * Set up the initial double-write page that lists all the buffers
+	 * that will be written to the double write file This list includes
+	 * the checksums of all the buffers and is checksummed itself.
+	 */
+	struct DoubleBufHeader *hdr = (struct DoubleBufHeader *) doubleBuf;
+	struct DoubleBufItem *item = hdr->items;
+
+	/*
+	 * The double-write header size should be big enough to contain info
+	 * for up to MAX_BATCH_BLOCKS buffers.
+	 */
+	Assert(sizeof(struct DoubleBufHeader) + MAX_BATCH_BLOCKS * sizeof(struct DoubleBufItem) <= DOUBLE_WRITE_HEADER_SIZE);
+
+	/* Remove any invalidated entries from the batch. */
+	for (i = 0; i < writeLen; ) {
+		if (writeList[i].forkNum == InvalidForkNumber)
+		{
+			elog(LOG, "Removing entry %d from dwbuf batch", i);
+			if (i < writeLen - 1) {
+				memcpy(&writeList[i], &writeList[i+1], sizeof(struct SMgrWriteList) * (writeLen - 1 -i));
+			}
+			writeLen--;
+		}
+		else
+		{
+			i++;
+		}
+	}
+
+	for (i = 0; i < writeLen; i++)
+	{
+		item->rnode = writeList[i].smgr_rnode;
+		item->forkNum = writeList[i].forkNum;
+		item->blockNum = writeList[i].blockNum;
+		item->slot = writeList[i].slot;
+		p = (PageHeader) writeList[i].buffer;
+		item->cksum = p->cksum;
+		item->pd_lsn = p->pd_lsn;
+		item++;
+
+		writeList[i].localrel = smgropen(writeList[i].smgr_rnode.node,
+										 writeList[i].smgr_rnode.backend);
+	}
+	hdr->writeLen = writeLen;
+	hdr->cksum = ComputeChecksum((uint64 *) hdr, DOUBLE_WRITE_HEADER_SIZE, 0);
+
+	/* XXX */
+	(*(smgrsw[0].smgr_bwrite)) (writeLen, writeList,
+												 doubleWriteFile, doubleBuf);
+	/* Zero out part of header that we filled in. */
+	memset(doubleBuf, 0,
+		   (char *) &(((struct DoubleBufHeader *) doubleBuf)->items[writeLen]) - doubleBuf);
 }
 
 /*
@@ -561,3 +1260,164 @@ smgrpostckpt(void)
 			(*(smgrsw[i].smgr_post_ckpt)) ();
 	}
 }
+
+extern char *double_write_directory;
+
+/*
+ * Return the name of the double write file.  Caller must use pfree().
+ */
+char *
+DoubleWriteFileName(int index)
+{
+	char	   *name;
+
+	if (double_write_directory != NULL)
+	{
+		name = palloc(strlen(double_write_directory) + strlen("/double") + 2 + 1);
+		sprintf(name, "%s/double.%d", double_write_directory, index);
+	}
+	else
+	{
+		name = palloc(strlen("base") + strlen("/double") + 2 + 1);
+		sprintf(name, "base/double.%d", index);
+	}
+	return name;
+}
+
+/*
+ * Called by postmaster at startup during recovery to read double-write
+ * file and see if there are any data blocks to be recovered.
+ */
+void
+RecoverDoubleWriteFile()
+{
+	char	   *name;
+	struct stat stat_buf;
+	File		fd;
+	int			r;
+	struct DoubleBufHeader *hdr;
+	int			i;
+	uint32		savedCksum;
+
+	name = DoubleWriteFileName(0);
+	if (stat(name, &stat_buf) == -1)
+	{
+		elog(LOG, "No double-write file");
+		pfree(name);
+		return;
+	}
+	fd = PathNameOpenFile(name, O_RDONLY, S_IRUSR | S_IWUSR);
+	if (fd < 0)
+	{
+		elog(PANIC, "Double-write file %s exists but can't be opened", name);
+	}
+
+	r = FileRead(fd, doubleBuf, DOUBLE_WRITE_HEADER_SIZE);
+	if (r < 0)
+		goto badformat;
+
+	hdr = (struct DoubleBufHeader *) doubleBuf;
+	if (hdr->writeLen == 0 && hdr->cksum == 0)
+	{
+		elog(LOG, "Double-write file %s is zeroed out", name);
+		FileClose(fd);
+		pfree(name);
+		return;
+	}
+	if (hdr->writeLen < 1 || hdr->writeLen > MAX_BATCH_BLOCKS)
+		goto badformat;
+
+	savedCksum = hdr->cksum;
+	hdr->cksum = 0;
+	if (savedCksum !=
+		ComputeChecksum((uint64 *) hdr, DOUBLE_WRITE_HEADER_SIZE, 0))
+		goto badformat;
+
+	for (i = 0; i < hdr->writeLen; i++)
+	{
+		struct DoubleBufItem *it = &(hdr->items[i]);
+		SMgrRelation smgr;
+		bool		mismatch;
+		PageHeader	p;
+
+		/*
+		 * For each block described in double-write file header, see if the
+		 * block in the database file has a checksum mismatch.	If so, restore
+		 * the block from the double-write file if that entry has correct
+		 * checksum.
+		 */
+		smgr = smgropen(it->rnode.node, it->rnode.backend);
+		mismatch = false;
+
+		/*
+		 * The block may no longer exist if relation was deleted/truncated
+		 * after the last double-write.
+		 */
+		if (!smgrexists(smgr, it->forkNum) ||
+			it->blockNum >= smgrnblocks(smgr, it->forkNum))
+		{
+			elog(LOG, "Block %s/%d in slot %d of double-write file no longer exists, skipping",
+				 relpath(it->rnode, it->forkNum), it->blockNum, i);
+			continue;
+		}
+
+		smgrread(smgr, it->forkNum, it->blockNum, (char *) tempbuf, &mismatch);
+		if (mismatch)
+		{
+			/*
+			 * The corresponding data block has a checksum error, and is
+			 * likely a torn page.	See if the block in the double-write file
+			 * has the correct checksum.  If so, we can correct the data page
+			 * from the block in the double-write file.
+			 */
+			FileSeek(fd, DOUBLE_WRITE_HEADER_SIZE + it->slot * BLCKSZ, SEEK_SET);
+			FileRead(fd, (char *) tempbuf, BLCKSZ);
+			p = (PageHeader) tempbuf;
+			savedCksum = p->cksum;
+			p->cksum = 0;
+			if (savedCksum != it->cksum ||
+				savedCksum != ComputeChecksum((uint64 *) tempbuf, BLCKSZ,
+											  ChecksumInit(smgr, it->forkNum,
+														   it->blockNum)))
+			{
+				elog(LOG, "Block %s/%d has checksum error, but can't be fixed, because slot %d of double-write file %s looks invalid",
+					 relpath(it->rnode, it->forkNum), it->blockNum, i, name);
+			}
+			else
+			{
+				/*
+				 * Correct the block in the data file from the block in the
+				 * double-write file.
+				 */
+				Assert(XLByteEQ(p->pd_lsn, it->pd_lsn));
+				smgrwrite(smgr, it->forkNum, it->blockNum, (char *) tempbuf, false, false);
+				elog(LOG, "Fixed block %s/%d (which had checksum error) from double-write file %s slot %d",
+					 relpath(it->rnode, it->forkNum), it->blockNum, name, i);
+			}
+		}
+		else
+		{
+			elog(DEBUG1, "Skipping slot %d of double-write file because block %s/%d is correct",
+				 i, relpath(it->rnode, it->forkNum), it->blockNum);
+		}
+		smgrclose(smgr);
+	}
+	FileClose(fd);
+
+	/*
+	 * Remove double-write file, so it can't be re-applied again if crash
+	 * happens during recovery.
+	 */
+	if (unlink(name) == -1)
+	{
+		elog(PANIC, "Failed to remove double-write file");
+	}
+	pfree(name);
+	return;
+
+badformat:
+	elog(LOG, "Double-write file %s has bad format", name);
+	FileClose(fd);
+	pfree(name);
+	return;
+}
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 7df5292..17c54aa 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -61,6 +61,7 @@
 #include "storage/bufmgr.h"
 #include "storage/standby.h"
 #include "storage/fd.h"
+#include "storage/ioseq.h"
 #include "storage/predicate.h"
 #include "tcop/tcopprot.h"
 #include "tsearch/ts_cache.h"
@@ -130,6 +131,9 @@ extern int	CommitSiblings;
 extern char *default_tablespace;
 extern char *temp_tablespaces;
 extern bool synchronize_seqscans;
+bool		doubleWrites;
+char	   *double_write_directory;
+extern bool fullPageWrites;
 extern int	ssl_renegotiation_limit;
 extern char *SSLCipherSuites;
 
@@ -419,6 +423,8 @@ bool		default_with_oids = false;
 bool		SQL_inheritance = true;
 
 bool		Password_encryption = true;
+bool		page_checksum = true;
+
 
 int			log_min_error_statement = ERROR;
 int			log_min_messages = WARNING;
@@ -445,6 +451,8 @@ int			tcp_keepalives_idle;
 int			tcp_keepalives_interval;
 int			tcp_keepalives_count;
 
+int			batched_buffer_writes = 0;
+
 /*
  * These variables are all dummies that don't do anything, except in some
  * cases provide the value for SHOW to display.  The real state is elsewhere
@@ -816,6 +824,14 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 	{
+		{"double_writes", PGC_POSTMASTER, DEVELOPER_OPTIONS,
+			gettext_noop("Page writes are written first to a temporary file before being written to the database file."),
+			NULL
+		},
+		&doubleWrites,
+		false, NULL, NULL
+	},
+	{
 		{"zero_damaged_pages", PGC_SUSET, DEVELOPER_OPTIONS,
 			gettext_noop("Continues processing past damaged page headers."),
 			gettext_noop("Detection of a damaged page header normally causes PostgreSQL to "
@@ -1438,6 +1454,23 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"page_checksum", PGC_POSTMASTER, CUSTOM_OPTIONS,
+			gettext_noop("enable disk page checksumming"),
+			NULL
+		},
+		&page_checksum, true, NULL, NULL
+	},
+
+	{
+		{"sort_dirty_buffers", PGC_POSTMASTER, DEVELOPER_OPTIONS,
+			gettext_noop("Sort dirty buffers before writing them out."),
+		    NULL,
+		    GUC_NOT_IN_SAMPLE
+		},
+		&sortDirtyBuffers, false, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL
@@ -2370,6 +2403,16 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"batched_buffer_writes", PGC_POSTMASTER, DEVELOPER_OPTIONS,
+			gettext_noop("Size of the block buffer for double writes."),
+			NULL,
+			GUC_NOT_IN_SAMPLE
+		},
+		&batched_buffer_writes,
+		64, 0, MAX_BATCH_BLOCKS, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
@@ -3018,6 +3061,16 @@ static struct config_string ConfigureNamesString[] =
 		check_application_name, assign_application_name, NULL
 	},
 
+	{
+		{"double_write_directory", PGC_POSTMASTER, DEVELOPER_OPTIONS,
+			gettext_noop("Location of the double write file."),
+			NULL,
+			GUC_REPORT | GUC_NOT_IN_SAMPLE
+		},
+		&double_write_directory,
+		NULL, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, NULL, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 400c52b..48ed1df 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -521,6 +521,7 @@
 # LOCK MANAGEMENT
 #------------------------------------------------------------------------------
 
+#page_checksum = on
 #deadlock_timeout = 1s
 #max_locks_per_transaction = 64		# min 10
 					# (change requires restart)
@@ -564,3 +565,6 @@
 #------------------------------------------------------------------------------
 
 # Add settings for extensions here
+double_writes = on
+#double_write_directory = ''
+#sort_dirty_buffers = off
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index fa38803..98aafc6 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -124,7 +124,7 @@ extern void simple_heap_update(Relation relation, ItemPointer otid,
 extern void heap_markpos(HeapScanDesc scan);
 extern void heap_restrpos(HeapScanDesc scan);
 
-extern void heap_sync(Relation relation);
+extern void heap_sync(Relation relation, bool needsDoubleWrite);
 
 extern void heap_redo(XLogRecPtr lsn, XLogRecord *rptr);
 extern void heap_desc(StringInfo buf, uint8 xl_info, char *rec);
diff --git a/src/include/postmaster/bgwriter.h b/src/include/postmaster/bgwriter.h
index 6cc4b62..5667f50 100644
--- a/src/include/postmaster/bgwriter.h
+++ b/src/include/postmaster/bgwriter.h
@@ -35,4 +35,6 @@ extern void AbsorbFsyncRequests(void);
 extern Size BgWriterShmemSize(void);
 extern void BgWriterShmemInit(void);
 
+bool AmCheckpointer(void);
+
 #endif   /* _BGWRITER_H */
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index de1bbd0..bba0594 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -183,7 +183,7 @@ extern void CheckPointBuffers(int flags);
 extern BlockNumber BufferGetBlockNumber(Buffer buffer);
 extern BlockNumber RelationGetNumberOfBlocksInFork(Relation relation,
 								ForkNumber forkNum);
-extern void FlushRelationBuffers(Relation rel);
+extern void FlushRelationBuffers(Relation rel, bool needsDoubleWrite);
 extern void FlushDatabaseBuffers(Oid dbid);
 extern void DropRelFileNodeBuffers(RelFileNodeBackend rnode,
 					   ForkNumber forkNum, BlockNumber firstDelBlock);
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index 1ab64e0..3dff3f7 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -132,6 +132,7 @@ typedef struct PageHeaderData
 	LocationIndex pd_special;	/* offset to start of special space */
 	uint16		pd_pagesize_version;
 	TransactionId pd_prune_xid; /* oldest prunable XID, or zero if none */
+	uint32		cksum;			/* page checksum */
 	ItemIdData	pd_linp[1];		/* beginning of line pointer array */
 } PageHeaderData;
 
@@ -165,8 +166,9 @@ typedef PageHeaderData *PageHeader;
  * Release 8.3 uses 4; it changed the HeapTupleHeader layout again, and
  *		added the pd_flags field (by stealing some bits from pd_tli),
  *		as well as adding the pd_prune_xid field (which enlarges the header).
+ * Release x.y uses 5; we added checksums to heap/index/fsm files.
  */
-#define PG_PAGE_LAYOUT_VERSION		4
+#define PG_PAGE_LAYOUT_VERSION		5
 
 
 /* ----------------------------------------------------------------
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index 22e7fe8..be5c3d3 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -53,11 +53,18 @@ typedef int File;
 /* GUC parameter */
 extern int	max_files_per_process;
 
+/*
+ * Size of the header on the double-write file that describes the buffers
+ * currently written to the double-write file.
+ */
+#define DOUBLE_WRITE_HEADER_SIZE 4096
 
 /*
  * prototypes for functions in fd.c
  */
 
+struct SMgrWriteList;
+
 /* Operations on virtual Files --- equivalent to Unix kernel file ops */
 extern File PathNameOpenFile(FileName fileName, int fileFlags, int fileMode);
 extern File OpenTemporaryFile(bool interXact);
@@ -67,6 +74,8 @@ extern int	FilePrefetch(File file, off_t offset, int amount);
 extern int	FileRead(File file, char *buffer, int amount);
 extern int	FileWrite(File file, char *buffer, int amount);
 extern int	FileSync(File file);
+extern int FileBwrite(int writeLen, struct SMgrWriteList *writeList,
+		   File doubleWriteFile, char *doubleBuf);
 extern off_t FileSeek(File file, off_t offset, int whence);
 extern int	FileTruncate(File file, off_t offset);
 extern char *FilePathName(File file);
diff --git a/src/include/storage/ioseq.h b/src/include/storage/ioseq.h
new file mode 100644
index 0000000..35efd4a
--- /dev/null
+++ b/src/include/storage/ioseq.h
@@ -0,0 +1,37 @@
+/* ************************************************************************
+ * Copyright 2010 VMware, Inc.  All rights reserved. -- VMware Confidential
+ * ************************************************************************/
+
+/*
+ *-------------------------------------------------------------------------
+ *
+ * ioseq.h
+ *	  Exports from storage/buffer/ioseq.c.
+ * 
+ * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef _IOSEQ_H
+#define _IOSEQ_H
+
+#include "storage/buf_internals.h"
+
+/* Structure used for sorting buffers on tag and eviction priority */
+typedef struct {
+	int idx;       /* Index in BufferDescriptors */
+	int clock;     /* Sequence number in clock order: clock(nextVictim) = 0 */
+	BufferTag tag; /* Tag of the buffer */
+} BufferIdxTag;
+
+extern bool sortDirtyBuffers;
+extern BufferIdxTag *ioSeqList;
+
+void IOSeq_Init(void);
+Size IOSeq_ShmemSize(void);
+void IOSeq_Sort(BufferIdxTag *entries, unsigned numEntries);
+unsigned IOSeq_ClockStart(BufferIdxTag *entries, unsigned numEntries);
+
+#endif   /* _IOSEQ_H */
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 46c8402..d3fc80b 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -15,8 +15,10 @@
 #define SMGR_H
 
 #include "fmgr.h"
+#include "fd.h"
 #include "storage/block.h"
 #include "storage/relfilenode.h"
+#include "access/xlogdefs.h"
 
 
 /*
@@ -71,6 +73,116 @@ typedef SMgrRelationData *SMgrRelation;
 #define SmgrIsTemp(smgr) \
 	((smgr)->smgr_rnode.backend != InvalidBackendId)
 
+/* Number of blocks in the double-write buffer */
+#define MAX_DW_BLOCKS		128
+#define CKPT_DW_BLOCKS		64
+#define NON_CKPT_DW_BLOCKS	128
+
+/* Maximum number of blocks in a batch written all at once. */
+/* XXX How to make sure we can cache all needed fds in Vfd cache */
+#define MAX_BATCH_BLOCKS 64
+
+/* Maximum number of batches per double-write buffer */
+#define MAX_BATCHES_PER_DWBUF  8
+
+/*
+ * Element of an array specifying a buffer write in a list of buffer
+ * writes being batched.
+ */
+struct SMgrWriteList
+{
+	RelFileNodeBackend smgr_rnode;
+	ForkNumber	forkNum;
+	BlockNumber blockNum;
+	int			slot;
+	char	   *buffer;
+
+	SMgrRelation localrel;
+
+	/* Filled in by mdbwrite */
+	File		fd;
+	off_t		seekPos;
+	int			len;
+};
+
+#include "storage/lwlock.h"
+
+struct DWBufferInfo
+{
+	char *bufferStart;
+	/* length of this double-write buffer in blocks, power of 2 */
+	int length;
+	/* mask for modding index to circular buffer, equals (length-1) */
+	int mask;
+
+	/*
+	 * start, endAlloc, endFill are monotonically increasing, must
+	 * alwayed be modded by MAX_DW_BLOCKS to access writeList.
+	 */
+	/* Start of allocated entries in writeList, advanced as batches are
+	 * flushed out. */
+	volatile int start;
+	/* End of allocated entries */
+	volatile int endAlloc;
+	/* End of entries that have been filled in */
+	volatile int endFill;
+
+	/*
+	 * Lock protecting writeList and start/endFill/flushed
+	 */
+	LWLockId writeLock;
+	LWLockId allocLock;
+	/* Has this batch been flushed? */
+	int flushed[MAX_DW_BLOCKS];
+	/* Lock per batch */
+	LWLockId batchLocks[MAX_DW_BLOCKS];
+	struct SMgrWriteList writeList[MAX_DW_BLOCKS];
+
+	int readHits;
+	int fullWaits;
+};
+
+/*
+ * One double-write buffer for all processes except the checkpointer, and
+ * one double-write buffer for the checkpointer.
+ */
+#define DWBUF_NON_CHECKPOINTER 0
+#define DWBUF_CHECKPOINTER 1
+
+/*
+ * Description of one buffer in the double-write file, as listed in the
+ * header.
+ */
+struct DoubleBufItem
+{
+	/* Specification of where this buffer is in the database */
+	RelFileNodeBackend rnode;	/* physical relation identifier */
+	ForkNumber	forkNum;
+	BlockNumber blockNum;		/* blknum relative to begin of reln */
+	int			slot;			/* Slot where buffer is stored */
+
+	/* Checksum of the buffer. */
+	int32		cksum;
+	/* LSN of the buffer */
+	XLogRecPtr	pd_lsn;
+};
+
+/* Format of the header of the double-write file */
+struct DoubleBufHeader
+{
+	uint32		cksum;
+	int32		writeLen;
+	uint32		pad1;
+	uint32		pad2;
+	struct DoubleBufItem items[0];
+};
+
+extern char *dwBlocks;
+extern struct DWBufferInfo *dwInfo;
+extern int batched_buffer_writes;
+extern bool doubleWrites;
+extern bool page_checksum;
+
 extern void smgrinit(void);
 extern SMgrRelation smgropen(RelFileNode rnode, BackendId backend);
 extern void smgrsettransient(SMgrRelation reln);
@@ -87,9 +199,12 @@ extern void smgrextend(SMgrRelation reln, ForkNumber forknum,
 extern void smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 			 BlockNumber blocknum);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
-		 BlockNumber blocknum, char *buffer);
+		 BlockNumber blocknum, char *buffer, bool *cksumMismatch);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,
-		  BlockNumber blocknum, char *buffer, bool skipFsync);
+                      BlockNumber blocknum, char *buffer, bool skipFsync,
+                      bool dw);
+extern void smgrbwrite(int writeLen, struct SMgrWriteList *writeList,
+		   File doubleWriteFile);
 extern BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum);
 extern void smgrtruncate(SMgrRelation reln, ForkNumber forknum,
 			 BlockNumber nblocks);
@@ -98,6 +213,10 @@ extern void smgrpreckpt(void);
 extern void smgrsync(void);
 extern void smgrpostckpt(void);
 
+extern void FlushDoubleWriteBuffer(int dwIndex);
+extern char *DoubleWriteFileName(int index);
+extern void RecoverDoubleWriteFile(void);
+
 
 /* internals: move me elsewhere -- ay 7/94 */
 
@@ -115,6 +234,8 @@ extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	   char *buffer);
 extern void mdwrite(SMgrRelation reln, ForkNumber forknum,
 		BlockNumber blocknum, char *buffer, bool skipFsync);
+extern void mdbwrite(int writeLen, struct SMgrWriteList *writeList,
+		 File doubleWriteFile, char *doublebuf);
 extern BlockNumber mdnblocks(SMgrRelation reln, ForkNumber forknum);
 extern void mdtruncate(SMgrRelation reln, ForkNumber forknum,
 		   BlockNumber nblocks);