Align inter-thread shared structers on cache-line granularity

Specifically ypipe_t specializations for msg_t and command_t.

Currently only supports platforms where posix_memalign is available.

src/yqueue.hpp

@@ -33,7 +33,7 @@ namespace zmq
     //  to minimise number of allocations/deallocations needed. Thus yqueue
     //  allocates/deallocates elements in batches of N.
     //
-    //  yqueue allows one thread to use push/back function and another one 
+    //  yqueue allows one thread to use push/back function and another one
     //  to use pop/front functions. However, user must ensure that there's no
     //  pop on the empty queue and that both threads don't access the same
     //  element in unsynchronised manner.
@@ -41,8 +41,16 @@ namespace zmq
     //  T is the type of the object in the queue.
     //  N is granularity of the queue (how many pushes have to be done till
     //  actual memory allocation is required).
-
+#ifdef HAVE_POSIX_MEMALIGN
+    // ALIGN is the memory alignment size to use in the case where we have
+    // posix_memalign available. Default value is 64, this alignment will
+    // prevent two queue chunks from occupying the same CPU cache line on
+    // architectures where cache lines are <= 64 bytes (e.g. most things
+    // except POWER).
+    template <typename T, int N, size_t ALIGN = 64> class yqueue_t
+#else
     template <typename T, int N> class yqueue_t
+#endif
     {
     public:
 
@@ -65,7 +73,7 @@ namespace zmq
                 if (begin_chunk == end_chunk) {
                     free (begin_chunk);
                     break;
-                } 
+                }
                 chunk_t *o = begin_chunk;
                 begin_chunk = begin_chunk->next;
                 free (o);
@@ -103,7 +111,13 @@ namespace zmq
                 end_chunk->next = sc;
                 sc->prev = end_chunk;
             } else {
+#ifdef HAVE_POSIX_MEMALIGN
+                void *pv;
+                if (posix_memalign(&pv, ALIGN, sizeof (chunk_t)) == 0)
+                    end_chunk->next = (chunk_t*) pv;
+#else
                 end_chunk->next = (chunk_t*) malloc (sizeof (chunk_t));
+#endif
                 alloc_assert (end_chunk->next);
                 end_chunk->next->prev = end_chunk;
             }