Re: smsd, postgresql and large numbers of messages - bugs?

[drbob]

On Sun, 27 Mar 2011 02:12:42 +0000, drbob wrote:

> 
> I'll try to actually test the code in the morning and report back here.
> 

So there were a couple of bugs and the patches in my previous post don't 
work

The patch below (against 0.6.26) works for me, I now get a separate 
transaction for each row of the outbox table instead of one big 
transaction that locks all the new rows.

Any comments welcome etc etc

--- pq.c        2007-05-07 19:05:51.000000000 +0100
+++ pq.c.new    2011-03-27 21:52:28.000000000 +0100
@@ -162,14 +162,13 @@
     g_string_sprintf (phnStr, "AND phone = '%s'", phone);
   }
 
-  buf = g_string_sized_new (128);
-
-  res1 = PQexec (connOut, "BEGIN");
-  PQclear (res1);
+  buf = g_string_sized_new (256);
 
+  gn_log_xdebug ("Performing initial SELECT\n");
+  
   g_string_sprintf (buf, "SELECT id, number, text, dreport FROM %s.outbox 
\
                           WHERE processed='f' AND localtime(0) >= 
not_before \
-                          AND localtime(0) <= not_after %s FOR UPDATE",
+                          AND localtime(0) <= not_after %s",
                           schema, phnStr->str);
   g_string_free (phnStr, TRUE);
 
@@ -179,16 +178,40 @@
     g_print (_("%d: SELECT FROM %s.outbox command failed.\n"), __LINE__, 
schema);
     g_print (_("Error: %s\n"), PQerrorMessage (connOut));
     PQclear (res1);
-    res1 = PQexec (connOut, "ROLLBACK TRANSACTION");
-    PQclear (res1);
     g_string_free (buf, TRUE);
     return;
   }
 
   for (i = 0; i < PQntuples (res1); i++)
   {
+    res2 = PQexec (connOut, "BEGIN");
+    if (!res2 || PQresultStatus (res2) != PGRES_COMMAND_OK)
+    {
+      g_print (_("%d: BEGIN command failed.\n"), __LINE__);
+      gn_log_xdebug ("%s\n", buf->str);
+      g_print (_("Error: %s\n"), PQerrorMessage (connOut));
+      PQclear(res2);
+    }
+    PQclear(res2);
+    g_string_printf (buf,"SELECT id, number, text, dreport FROM %s.outbox 
\
+                         WHERE processed='f' AND id=%s AND localtime(0) 
>= not_before \
+                         AND localtime(0) <= not_after FOR UPDATE NOWAIT",
+                         schema,PQgetvalue (res1, i, 0));
+
+    res2 = PQexec (connOut, buf->str);
+    if (!res2 || PQresultStatus (res2) != PGRES_TUPLES_OK || PQntuples
(res2) == 0)
+    {
+      g_print (_("%d: SELECT FOR UPDATE (row lock) command failed.\n"), 
__LINE__);
+      gn_log_xdebug ("%s\n", buf->str);
+      g_print (_("Error: %s\n"), PQerrorMessage (connOut));
+      PQclear(res2);
+      res2 = PQexec (connOut, "ROLLBACK TRANSACTION");
+      PQclear (res2);
+      continue;
+    }
+
     gn_sms sms;
-    
+   
     gn_sms_default_submit (&sms);
     memset (&sms.remote.number, 0, sizeof (sms.remote.number));
     sms.delivery_report = atoi (PQgetvalue (res1, i, 3));
@@ -226,16 +249,23 @@
     if (!res2 || PQresultStatus (res2) != PGRES_COMMAND_OK)
     {
       g_print (_("%d: UPDATE command failed.\n"), __LINE__);   
+      gn_log_xdebug ("%s\n", buf->str);      
       g_print (_("Error: %s\n"), PQerrorMessage (connOut));
     }
 
     PQclear (res2);
+
+    res2 = PQexec (connOut, "COMMIT");
+    if (!res2 || PQresultStatus (res2) != PGRES_COMMAND_OK)
+    {
+      g_print (_("%d: COMMIT command failed.\n"), __LINE__);
+      gn_log_xdebug ("%s\n", buf->str);
+      g_print (_("Error: %s\n"), PQerrorMessage (connOut));
+    }
+    PQclear(res2);
   }
 
   PQclear (res1);
-
-  res1 = PQexec(connOut, "COMMIT");
-
   g_string_free(buf, TRUE);
-  PQclear (res1);
-}
+
+}
\ No newline at end of file