changeset 835:1d439494604c

Memory barriers in ringbuffer

--- a/base/RingBuffer.h	Tue Jul 30 18:15:30 2013 +0100
+++ b/base/RingBuffer.h	Mon Sep 16 15:47:27 2013 +0100
@@ -24,7 +24,6 @@
 #include <sys/types.h>
 
 #include "system/System.h"
-#include "Scavenger.h"
 
 #include <cstring> // memcpy, memset &c
 
@@ -67,11 +66,15 @@
     size_t getSize() const;
 
     /**
-     * Resize the ring buffer.  This also empties it.  Actually swaps
-     * in a new, larger buffer; the old buffer is scavenged after a
-     * seemly delay.  Should be called from the write thread.
+     * Return a new ring buffer (allocated with "new" -- caller must
+     * delete when no longer needed) of the given size, containing the
+     * same data as this one as perceived by reader 0 of this buffer.
+     * If another thread reads from or writes to this buffer during
+     * the call, the contents of the new buffer may be incomplete or
+     * inconsistent.  If this buffer's data will not fit in the new
+     * size, the contents are undefined.
      */
-    void resize(size_t newSize);
+    RingBuffer<T, N> *resized(size_t newSize) const;
 
     /**
      * Lock the ring buffer into physical memory.  Returns true
@@ -167,17 +170,12 @@
     size_t  m_size;
     size_t  m_spare;
 
-    static Scavenger<ScavengerArrayWrapper<T> > m_scavenger;
-
 private:
     RingBuffer(const RingBuffer &); // not provided
     RingBuffer &operator=(const RingBuffer &); // not provided
 };
 
 template <typename T, int N>
-Scavenger<ScavengerArrayWrapper<T> > RingBuffer<T, N>::m_scavenger;
-
-template <typename T, int N>
 RingBuffer<T, N>::RingBuffer(size_t n) :
     m_buffer(new T[n + 1]),
     m_mlocked(false),
@@ -200,8 +198,6 @@
 */
     
     for (int i = 0; i < N; ++i) m_readers[i] = 0;
-
-    m_scavenger.scavenge();
 }
 
 template <typename T, int N>
@@ -217,8 +213,6 @@
 	MUNLOCK((void *)m_buffer, m_size * sizeof(T));
     }
     delete[] m_buffer;
-
-    m_scavenger.scavenge();
 }
 
 template <typename T, int N>
@@ -233,30 +227,25 @@
 }
 
 template <typename T, int N>
-void
-RingBuffer<T, N>::resize(size_t newSize)
+RingBuffer<T, N> *
+RingBuffer<T, N>::resized(size_t newSize) const
 {
 #ifdef DEBUG_RINGBUFFER
-    std::cerr << "RingBuffer<T," << N << ">[" << this << "]::resize(" << newSize << ")" << std::endl;
+    std::cerr << "RingBuffer<T," << N << ">[" << this << "]::resized(" << newSize << ")" << std::endl;
 #endif
 
-    m_scavenger.scavenge();
+    RingBuffer<T, N> *newBuffer = new RingBuffer<T, N>(newSize);
 
-    if (m_mlocked) {
-	MUNLOCK((void *)m_buffer, m_size * sizeof(T));
+    int w = m_writer;
+    int r = m_readers[0];
+
+    while (r != w) {
+        T value = m_buffer[r];
+        newBuffer->write(&value, 1);
+        if (++r == m_size) r = 0;
     }
 
-    m_scavenger.claim(new ScavengerArrayWrapper<T>(m_buffer));
-
-    reset();
-    m_buffer = new T[newSize + 1];
-    m_size = newSize + 1;
-
-    if (m_mlocked) {
-	if (MLOCK((void *)m_buffer, m_size * sizeof(T))) {
-	    m_mlocked = false;
-	}
-    }
+    return newBuffer;
 }
 
 template <typename T, int N>
@@ -350,6 +339,7 @@
 	memcpy(destination + here, m_buffer, (n - here) * sizeof(T));
     }
 
+    MBARRIER();
     m_readers[R] = (m_readers[R] + n) % m_size;
 
 #ifdef DEBUG_RINGBUFFER
@@ -392,6 +382,7 @@
 	}
     }
 
+    MBARRIER();
     m_readers[R] = (m_readers[R] + n) % m_size;
     return n;
 }
@@ -414,6 +405,7 @@
 	return t;
     }
     T value = m_buffer[m_readers[R]];
+    MBARRIER();
     if (++m_readers[R] == m_size) m_readers[R] = 0;
     return value;
 }
@@ -520,6 +512,7 @@
 	memcpy(m_buffer, source + here, (n - here) * sizeof(T));
     }
 
+    MBARRIER();
     m_writer = (m_writer + n) % m_size;
 
 #ifdef DEBUG_RINGBUFFER
@@ -554,7 +547,8 @@
 	memset(m_buffer + m_writer, 0, here * sizeof(T));
 	memset(m_buffer, 0, (n - here) * sizeof(T));
     }
-
+    
+    MBARRIER();
     m_writer = (m_writer + n) % m_size;
     return n;
 }

--- a/system/System.cpp	Tue Jul 30 18:15:30 2013 +0100
+++ b/system/System.cpp	Mon Sep 16 15:47:27 2013 +0100
@@ -274,6 +274,30 @@
 #endif
 }
 
+#ifdef _WIN32
+extern void SystemMemoryBarrier()
+{
+#ifdef __MSVC__
+    MemoryBarrier();
+#else /* mingw */
+    LONG Barrier = 0;
+    __asm__ __volatile__("xchgl %%eax,%0 "
+                         : "=r" (Barrier));
+#endif
+}
+#else /* !_WIN32 */
+#if !defined(__APPLE__) && ((__GNUC__ < 4) || (__GNUC__ == 4 && __GNUC_MINOR__ == 0))
+void
+SystemMemoryBarrier()
+{
+    pthread_mutex_t dummy = PTHREAD_MUTEX_INITIALIZER;
+    pthread_mutex_lock(&dummy);
+    pthread_mutex_unlock(&dummy);
+}
+#endif /* !defined(__APPLE__) etc */
+#endif /* !_WIN32 */
+
+
 static char *startupLocale = 0;
 
 void

--- a/system/System.h	Tue Jul 30 18:15:30 2013 +0100
+++ b/system/System.h	Mon Sep 16 15:47:27 2013 +0100
@@ -30,6 +30,9 @@
 #define MUNLOCK_SAMPLEBLOCK(a) 1
 #define MUNLOCKALL() 1
 
+extern void SystemMemoryBarrier();
+#define MBARRIER()   SystemMemoryBarrier()
+
 #define DLOPEN(a,b)  LoadLibrary((a).toStdWString().c_str())
 #define DLSYM(a,b)   GetProcAddress((HINSTANCE)(a),(b))
 #define DLCLOSE(a)   (!FreeLibrary((HINSTANCE)(a)))
@@ -99,6 +102,9 @@
 
 #define MUNLOCKALL() 1
 
+#include <libkern/OSAtomic.h>
+#define MBARRIER() OSMemoryBarrier()
+
 #else 
 
 #ifdef sun
@@ -122,7 +128,14 @@
 
 #define MUNLOCKALL() ::munlockall()
 
-#endif /* __APPLE__ */
+#if (__GNUC__ > 4) || (__GNUC__ == 4 && __GNUC_MINOR__ >= 1)
+#define MBARRIER() __sync_synchronize()
+#else
+extern void SystemMemoryBarrier();
+#define MBARRIER() SystemMemoryBarrier()
+#endif
+
+#endif /* ! __APPLE__ */
 
 #endif /* ! _WIN32 */