Re: parallel restore fixes - Mailing list pgsql-hackers

Tom Lane wrote:
>
> How about this: by default, fmtId uses the same logic as now (one static
> PQExpBuffer).  If told to by a call of init_parallel_dump_utils(), which
> need only be called by pg_restore during its startup, then it switches to
> using per-thread storage.  init_parallel_dump_utils can be the place
> that calls TlsAlloc.  This is almost the same as what you suggested a
> couple messages back, but perhaps a bit clearer as to what's going on;
> and it definitely cuts the number of places we need to touch.
>
>
>

OK, here 'tis.

Moving on to the deadlock with crossed FKs issue.

cheers

andrew
? src/bin/pg_dump/.deps
? src/bin/pg_dump/kwlookup.c
? src/bin/pg_dump/win32ver.rc
Index: src/bin/pg_dump/dumputils.c
===================================================================
RCS file: /home/cvsmirror/pg/pgsql/src/bin/pg_dump/dumputils.c,v
retrieving revision 1.44
diff -c -r1.44 dumputils.c
*** src/bin/pg_dump/dumputils.c    22 Jan 2009 20:16:07 -0000    1.44
--- src/bin/pg_dump/dumputils.c    10 Mar 2009 19:38:07 -0000
***************
*** 31,55 ****
  static void AddAcl(PQExpBuffer aclbuf, const char *keyword,
                     const char *subname);


  /*
!  *    Quotes input string if it's not a legitimate SQL identifier as-is.
   *
!  *    Note that the returned string must be used before calling fmtId again,
!  *    since we re-use the same return buffer each time.  Non-reentrant but
!  *    avoids memory leakage.
   */
  const char *
  fmtId(const char *rawid)
  {
!     static PQExpBuffer id_return = NULL;
      const char *cp;
      bool        need_quotes = false;

      if (id_return)                /* first time through? */
          resetPQExpBuffer(id_return);
      else
          id_return = createPQExpBuffer();

      /*
       * These checks need to match the identifier production in scan.l. Don't
--- 31,102 ----
  static void AddAcl(PQExpBuffer aclbuf, const char *keyword,
                     const char *subname);

+ #ifdef WIN32
+ static bool parallel_init_done = false;
+ static DWORD tls_index;
+ #endif
+
+ void
+ init_parallel_dump_utils(void)
+ {
+ #ifdef WIN32
+     if (! parallel_init_done)
+     {
+         tls_index = TlsAlloc();
+         parallel_init_done = true;
+     }
+ #endif
+ }

  /*
!  *  Quotes input string if it's not a legitimate SQL identifier as-is.
   *
!  *  Note that the returned string must be used before calling fmtId again,
!  *  since we re-use the same return buffer each time.  Non-reentrant but
!  *  reduces memory leakage. (On Windows the memory leakage will be one buffer
!  *  per thread, which is at least better than one per call).
   */
  const char *
  fmtId(const char *rawid)
  {
!     /*
!      * The Tls code goes awry if we use a static var, so we provide for both
!      * static and auto, and omit any use of the static var when using Tls.
!      */
!     static PQExpBuffer s_id_return = NULL;
!     PQExpBuffer id_return;
!
      const char *cp;
      bool        need_quotes = false;

+ #ifdef WIN32
+     if (parallel_init_done)
+         id_return = (PQExpBuffer) TlsGetValue(tls_index); /* 0 when not set */
+     else
+         id_return = s_id_return;
+ #else
+     id_return = s_id_return;
+ #endif
+
      if (id_return)                /* first time through? */
+     {
+         /* same buffer, just wipe contents */
          resetPQExpBuffer(id_return);
+     }
      else
+     {
+         /* new buffer */
          id_return = createPQExpBuffer();
+ #ifdef WIN32
+         if (parallel_init_done)
+             TlsSetValue(tls_index,id_return);
+         else
+             s_id_return = id_return;
+ #else
+         s_id_return = id_return;
+ #endif
+
+     }

      /*
       * These checks need to match the identifier production in scan.l. Don't
Index: src/bin/pg_dump/dumputils.h
===================================================================
RCS file: /home/cvsmirror/pg/pgsql/src/bin/pg_dump/dumputils.h,v
retrieving revision 1.23
diff -c -r1.23 dumputils.h
*** src/bin/pg_dump/dumputils.h    22 Jan 2009 20:16:07 -0000    1.23
--- src/bin/pg_dump/dumputils.h    10 Mar 2009 19:38:07 -0000
***************
*** 19,24 ****
--- 19,25 ----
  #include "libpq-fe.h"
  #include "pqexpbuffer.h"

+ extern void init_parallel_dump_utils(void);
  extern const char *fmtId(const char *identifier);
  extern void appendStringLiteral(PQExpBuffer buf, const char *str,
                      int encoding, bool std_strings);
Index: src/bin/pg_dump/pg_backup_archiver.c
===================================================================
RCS file: /home/cvsmirror/pg/pgsql/src/bin/pg_dump/pg_backup_archiver.c,v
retrieving revision 1.165
diff -c -r1.165 pg_backup_archiver.c
*** src/bin/pg_dump/pg_backup_archiver.c    5 Mar 2009 14:51:10 -0000    1.165
--- src/bin/pg_dump/pg_backup_archiver.c    10 Mar 2009 19:38:07 -0000
***************
*** 3467,3478 ****

      /*
       * Close and reopen the input file so we have a private file pointer
!      * that doesn't stomp on anyone else's file pointer.
       *
!      * Note: on Windows, since we are using threads not processes, this
!      * *doesn't* close the original file pointer but just open a new one.
       */
!     (AH->ReopenPtr) (AH);

      /*
       * We need our own database connection, too
--- 3467,3486 ----

      /*
       * Close and reopen the input file so we have a private file pointer
!      * that doesn't stomp on anyone else's file pointer, if we're actually
!      * going to need to read from the file. Otherwise, just close it
!      * except on Windows, where it will possibly be needed by other threads.
       *
!      * Note: on Windows, since we are using threads not processes, the
!      * reopen call *doesn't* close the original file pointer but just open
!      * a new one.
       */
!     if (te->section == SECTION_DATA )
!         (AH->ReopenPtr) (AH);
! #ifndef WIN32
!     else
!         (AH->ClosePtr) (AH);
! #endif

      /*
       * We need our own database connection, too
***************
*** 3490,3496 ****
      PQfinish(AH->connection);
      AH->connection = NULL;

!     (AH->ClosePtr) (AH);

      if (retval == 0 && AH->public.n_errors)
          retval = WORKER_IGNORED_ERRORS;
--- 3498,3506 ----
      PQfinish(AH->connection);
      AH->connection = NULL;

!     /* If we reopened the file, we are done with it, so close it now */
!     if (te->section == SECTION_DATA )
!         (AH->ClosePtr) (AH);

      if (retval == 0 && AH->public.n_errors)
          retval = WORKER_IGNORED_ERRORS;
Index: src/bin/pg_dump/pg_restore.c
===================================================================
RCS file: /home/cvsmirror/pg/pgsql/src/bin/pg_dump/pg_restore.c,v
retrieving revision 1.94
diff -c -r1.94 pg_restore.c
*** src/bin/pg_dump/pg_restore.c    26 Feb 2009 16:02:38 -0000    1.94
--- src/bin/pg_dump/pg_restore.c    10 Mar 2009 19:38:07 -0000
***************
*** 40,45 ****
--- 40,46 ----
   */

  #include "pg_backup_archiver.h"
+ #include "dumputils.h"

  #include <ctype.h>

***************
*** 125,130 ****
--- 126,133 ----

      set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_dump"));

+     init_parallel_dump_utils();
+
      opts = NewRestoreOptions();

      progname = get_progname(argv[0]);