Use Python actual tuple's to represent element3.Tuple objects.

This is done to avoid additional object allocations while processing tuple data.
Much of the overhead involved with parsing the tuple data ended up being due to
the additional objects being allocated while processing D messages.

Additionally, increase the default chunksize aggressively.
(Might tune this down)

Author: James William Pye

postgresql/driver/pq3.py

@@ -377,10 +377,14 @@ def _init(self,
 			self._pq_xp_fetchall() + (element.SynchronizeMessage,)
 		)
 		self.database._pq_push(self._xact, self)
+		STEP = self.database._pq_step
 		while self._xact.state != xact.Complete:
-			self.database._pq_step()
+			STEP()
 			for x in self._xact.messages_received():
-				if x.type == null:
+				if x.__class__ is tuple or expect == x.type:
+					# no need to step once this is seen
+					return
+				elif x.type == null:
 					self.database._pq_complete()
 					self._xact = None
 					return
@@ -392,9 +396,6 @@ def _init(self,
 					# return.
 					self._xact = None
 					return
-				elif x.type == expect:
-					# no need to step once this is seen
-					return
 				elif x.type in (bindcomplete, parsecomplete):
 					pass
 				else:
@@ -446,13 +447,13 @@ class SingleXactFetch(FetchAll):
 	_expect = element.Tuple.type
 	_process_chunk_ = FetchAll._process_tuple_chunk_Row
 
-	def _process_chunk(self, x, tuple_type = element.Tuple.type):
+	def _process_chunk(self, x, tuple_type = tuple):
 		return self._process_chunk_((
-			y for y in x if y.type == tuple_type
+			y for y in x if y.__class__ is tuple
 		))
 
 class MultiXactStream(Chunks):
-	chunksize = 512
+	chunksize = 1024 * 10
 	# only tuple streams
 	_process_chunk = Output._process_tuple_chunk_Row
 
@@ -483,7 +484,7 @@ def _init(self):
 		self._xact = self._ins(self._bind() + self._command)
 		self.database._pq_push(self._xact, self)
 
-	def __next__(self, tuple_type = element.Tuple.type):
+	def __next__(self, tuple_type = tuple):
 		x = self._xact
 		if x is None:
 			raise StopIteration
@@ -493,7 +494,7 @@ def __next__(self, tuple_type = element.Tuple.type):
 
 		# get all the element.Tuple messages
 		chunk = [
-			y for y in x.messages_received() if y.type == tuple_type
+			y for y in x.messages_received() if y.__class__ is tuple_type
 		]
 		if len(chunk) == self.chunksize:
 			# there may be more, dispatch the request for the next chunk
@@ -625,7 +626,7 @@ def _fetch(self, direction, quantity):
 		self.database._pq_push(x, self)
 		self.database._pq_complete()
 		return self._process_tuple((
-			y for y in x.messages_received() if y.type == element.Tuple.type
+			y for y in x.messages_received() if y.__class__ is tuple
 		))
 
 	def seek(self, offset, whence = 'ABSOLUTE'):
@@ -1093,7 +1094,7 @@ def first(self, *parameters):
 			# It returned rows, look for the first tuple.
 			tuple_type = element.Tuple.type
 			for xt in x.messages_received():
-				if xt.type == tuple_type:
+				if xt.__class__ is tuple:
 					break
 			else:
 				return None

postgresql/port/optimized/client3.c

@@ -11,6 +11,7 @@
 static PyObject *
 cat_messages(PyObject *self, PyObject *messages_in)
 {
+	const static uint32_t null_attribute = 0xFFFFFFFFL;
 	PyObject *msgs = NULL;
 	Py_ssize_t nmsgs = 0;
 	Py_ssize_t cmsg = 0;
@@ -62,7 +63,8 @@ cat_messages(PyObject *self, PyObject *messages_in)
 			 */
 
 			/* realloc the buf for the new copy data */
-			bufsize = bufsize + (5 * (eofc - cmsg)) + xsize;
+			xsize = xsize + (5 * (eofc - cmsg));
+			bufsize = bufsize + xsize;
 			nbuf = realloc(buf, bufsize);
 			if (nbuf == NULL)
 			{
@@ -99,6 +101,164 @@ cat_messages(PyObject *self, PyObject *messages_in)
 				++cmsg;
 			}
 		}
+		else if (PyTuple_CheckExact(ob))
+		{
+			/*
+			 * Handle 'D' tuple data from a raw Python tuple.
+			 */
+			Py_ssize_t eofc = cmsg;
+			Py_ssize_t xsize = 0;
+
+			/* find the last of the tuple data (eofc) */
+			do
+			{
+				Py_ssize_t current_item, nitems;
+
+				nitems = PyTuple_GET_SIZE(ob);
+				if (nitems > 0xFFFF)
+				{
+					PyErr_SetString(PyExc_OverflowError,
+						"too many attributes in tuple message");
+					goto fail;
+				}
+
+				/*
+				 * The items take *at least* 4 bytes each.
+				 * (The attribute count is considered later)
+				 */
+				xsize = xsize + (nitems * 4);
+
+				for (current_item = 0; current_item < nitems; ++current_item)
+				{
+					PyObject *att = PyTuple_GET_ITEM(ob, current_item);
+
+					/*
+					 * Attributes are expected to be bytes() or None.
+					 */
+					if (PyBytes_CheckExact(att))
+						xsize = xsize + PyBytes_GET_SIZE(att);
+					else if (att != Py_None)
+					{
+						PyErr_Format(PyExc_TypeError,
+							"cannot serialize tuple message attribute of type '%s'",
+							Py_TYPE(att)->tp_name);
+						goto fail;
+					}
+					/*
+					 * else it's Py_None and the size has been specified.
+					 */
+				}
+
+				++eofc;
+				if (eofc >= nmsgs)
+					break; /* end of messages in the list? */
+
+				/* Grab the next message. */
+				ob = PyList_GET_ITEM(msgs, eofc);
+			} while(PyTuple_CheckExact(ob));
+
+			/*
+			 * Either the end of the list or `ob` is not a data object meaning
+			 * that it's the end of the copy data.
+			 */
+
+			/*
+			 * realloc the buf for the new tuple data
+			 *
+			 * Each D message consumes at least 1 + 4 + 2 bytes:
+			 *  1 for the message type
+			 *  4 for the message size
+			 *  2 for the attribute count
+			 */
+			xsize = xsize + (7 * (eofc - cmsg));
+			bufsize = bufsize + xsize;
+			nbuf = realloc(buf, bufsize);
+			if (nbuf == NULL)
+			{
+				PyErr_Format(
+					PyExc_MemoryError,
+					"failed to allocate %lu bytes of memory for out-going messages",
+					(unsigned long) bufsize
+				);
+				goto fail;
+			}
+			else
+			{
+				buf = nbuf;
+				nbuf = NULL;
+			}
+
+			/*
+			 * Make the final pass through the tuple data memcpy'ing the data from
+			 * the bytes() objects.
+			 *
+			 * No type checks are done here as they should have been done while
+			 * gathering the sizes for the realloc().
+			 */
+			while (cmsg < eofc)
+			{
+				Py_ssize_t current_item, nitems;
+				uint32_t msg_length, out_msg_len;
+				uint16_t natts;
+				char *localbuf = buf + bufpos + 5; /* skipping the header for now */
+				buf[bufpos] = 'D'; /* Tuple data message type */
+
+				ob = PyList_GET_ITEM(msgs, cmsg);
+				nitems = PyTuple_GET_SIZE(ob);
+
+				/*
+				 * 4 bytes for the message length,
+				 * 2 bytes for the attribute count and
+				 * 4 bytes for each item in 'ob'.
+				 */
+				msg_length = 4 + 2 + (nitems * 4);
+
+				/*
+				 * Set number of attributes.
+				 */
+				natts = local_ntohs((uint16_t) nitems);
+				Py_MEMCPY(localbuf, &natts, 2);
+				localbuf = localbuf + 2;
+
+				for (current_item = 0; current_item < nitems; ++current_item)
+				{
+					PyObject *att = PyTuple_GET_ITEM(ob, current_item);
+
+					if (att == Py_None)
+					{
+						Py_MEMCPY(localbuf, &null_attribute, 4);
+						localbuf = localbuf + 4;
+					}
+					else
+					{
+						Py_ssize_t attsize = PyBytes_GET_SIZE(att);
+						uint32_t n_attsize;
+
+						n_attsize = local_ntohl((uint32_t) attsize);
+
+						Py_MEMCPY(localbuf, &n_attsize, 4);
+						localbuf = localbuf + 4;
+						Py_MEMCPY(localbuf, PyBytes_AS_STRING(att), attsize);
+						localbuf = localbuf + attsize;
+
+						msg_length = msg_length + attsize;
+					}
+				}
+
+				/*
+				 * Summed up the message size while copying the attributes.
+				 */
+				out_msg_len = local_ntohl(msg_length);
+				Py_MEMCPY(buf + bufpos + 1, &out_msg_len, 4);
+
+				/*
+				 * Filled in the data while summing the message size, so
+				 * adjust the buffer position for the next message.
+				 */
+				bufpos = bufpos + 1 + msg_length;
+				++cmsg;
+			}
+		}
 		else
 		{
 			PyObject *serialized;

postgresql/port/optimized/element3.c

@@ -112,12 +112,12 @@ _pack_tuple_data(PyObject *tup)
  * dst must be of PyTuple_Type with at least natts items slots.
  */
 static int
-_unpack_tuple_data(PyObject *dst, uint16_t natts, const char *data, Py_ssize_t data_len)
+_unpack_tuple_data(PyObject *dst, uint16_t natts, register const char *data, Py_ssize_t data_len)
 {
 	PyObject *ob;
 	uint16_t cnatt = 0;
 	uint32_t attsize = 0;
-	uint32_t position = 0;
+	register uint32_t position = 0;
 
 	while (cnatt < natts)
 	{
@@ -203,7 +203,7 @@ _unpack_tuple_data(PyObject *dst, uint16_t natts, const char *data, Py_ssize_t d
 static PyObject *
 parse_tuple_message(PyObject *self, PyObject *args)
 {
-	PyObject *prerob, *rob, *temp_tup;
+	PyObject *rob;
 	PyObject *typ;
 	const char *data;
 	Py_ssize_t dlen = 0;
@@ -221,26 +221,16 @@ parse_tuple_message(PyObject *self, PyObject *args)
 	Py_MEMCPY(&natts, data, 2);
 	natts = local_ntohs(natts);
 
-	prerob = PyTuple_New(natts);
-	if (prerob == NULL)
+	rob = PyTuple_New(natts);
+	if (rob == NULL)
 		return(NULL);
 
-	if (_unpack_tuple_data(prerob, natts, data+2, dlen-2) < 0)
-	{
-		Py_DECREF(prerob);
-		return(NULL);
-	}
-
-	temp_tup = PyTuple_New(1);
-	if (temp_tup == NULL)
+	if (_unpack_tuple_data(rob, natts, data+2, dlen-2) < 0)
 	{
-		Py_DECREF(prerob);
+		Py_DECREF(rob);
 		return(NULL);
 	}
-	PyTuple_SET_ITEM(temp_tup, 0, prerob);
 
-	rob = PyObject_CallObject(typ, temp_tup);
-	Py_DECREF(temp_tup);
 	return(rob);
 }
 

postgresql/protocol/client3.py

@@ -12,18 +12,6 @@
 
 __all__ = ('Connection',)
 
-try:
-	from ..port.optimized import cat_messages
-except ImportError:
-	from ..python.structlib import long_pack
-	def cat_messages(messages, lpack = long_pack, blen = bytes.__len__):
-		return b''.join([
-			x.bytes() if x.__class__ is not bytes else (
-				b'd' + lpack(blen(x) + 4) + x
-			) for x in messages
-		])
-	del long_pack
-
 client_detected_protocol_error = element.ClientError((
 	(b'S', 'FATAL'),
 	(b'C', '08P01'),
@@ -330,7 +318,9 @@ def send_message_data(self):
 				raise
 		return True
 
-	def standard_write_messages(self, messages):
+	def standard_write_messages(self, messages,
+		cat_messages = element.cat_messages
+	):
 		'protocol message writer'
 		if self.writing is not self.written:
 			self.message_data += cat_messages(self.writing)