Rewrite the atomics code to be simpler, cleaner and more portable.

Author: flanglet

src/concurrent.hpp

@@ -22,6 +22,7 @@ limitations under the License.
 #if __cplusplus >= 201103L || (defined(_MSC_VER) && _MSC_VER >= 1700)
     // C++ 11 (or partial)
     #include <atomic>
+    #define HAVE_STD_ATOMICS 1
 
     #ifndef CONCURRENCY_DISABLED
         #ifdef __clang__
@@ -37,7 +38,7 @@ limitations under the License.
         #endif
     #endif
 #else
-    #define CONCURRENT_FALLBACK_ATOMICS
+    #define HAVE_STD_ATOMICS 0
 #endif
 
 
@@ -204,171 +205,51 @@ class Task {
 #endif
 
 
-#ifdef CONCURRENT_FALLBACK_ATOMICS
-   // ! Stubs for NON CONCURRENT USAGE !
-   // Used when compiling for older C++ standards (C++98/03)
+#if HAVE_STD_ATOMICS
+    typedef std::atomic_int atomic_int_t;
 
-/*
-   // Use enum instead of const int to prevent linkage issues
-   enum fallback_memory_order {
-       memory_order_relaxed = 0,
-       memory_order_consume = 1,
-       memory_order_acquire = 2,
-       memory_order_release = 3,
-       memory_order_acq_rel = 4,
-       memory_order_seq_cst = 5
-   };
-*/
-
-
-   // Naming the class 'fallback_...' avoids conflicts if the macro
-   // matches the class name recursively in some preprocessors.
-   class fallback_atomic_int {
-   private:
-       volatile int _n;
-
-       // Disable copy constructor and assignment operator
-       // (Atomics should not be copyable)
-       fallback_atomic_int(const fallback_atomic_int&);
-       fallback_atomic_int& operator=(const fallback_atomic_int&);
-
-   public:
-       fallback_atomic_int(int n = 0) : _n(n) {}
-
-       // Assignment returns the value (int), NOT the object reference
-       int operator=(int n)
-       {
-           _n = n;
-           return n;
-       }
-
-       operator int() const
-       {
-           return _n;
-       }
-
-       int load(int mo = 0 /*memory_order_relaxed*/) const
-       {
-           (void)mo;
-           return _n;
-       }
-
-       void store(int n, int mo = 3 /*memory_order_release*/)
-       {
-           (void)mo;
-           _n = n;
-       }
-
-       // Postfix ++ (x++) returns OLD value
-       int operator++(int)
-       {
-           int old = _n;
-           _n++;
-           return old;
-       }
-
-       // Prefix ++ (++x) returns NEW value
-       int operator++()
-       {
-           return ++_n;
-       }
-
-       // Standard signature: takes int delta, returns OLD value
-       int fetch_add(int delta, int mo = 5 /*memory_order_seq_cst*/)
-       {
-          (void)mo;
-          int old = _n;
-          _n += delta;
-          return old;
-       }
-
-       int fetch_sub(int delta, int mo = 5 /*memory_order_seq_cst*/)
-       {
-          (void)mo;
-          int old = _n;
-          _n -= delta;
-          return old;
-       }
+    #define LOAD_ATOMIC(a)  ((a).load(std::memory_order_acquire))
+    #define STORE_ATOMIC(a, v) ((a).store((v), std::memory_order_release))
+    #define EXCHANGE_ATOMIC(a, v)  ((a).exchange((v), std::memory_order_acq_rel))
+    #define FETCH_ADD_ATOMIC(a, v) ((a).fetch_add((v), std::memory_order_acq_rel))
+    #define COMPARE_EXCHANGE_ATOMIC(obj, expected, desired) \
+                                    ((obj).compare_exchange_strong((expected), (desired), \
+                                    std::memory_order_release, std::memory_order_acquire))
 
-       // CRITICAL FIX: Must update 'expected' on failure
-       bool compare_exchange_strong(int& expected, int desired, int mo = 5 /*memory_order_seq_cst*/)
-       {
-           (void)mo;
-
-           if (_n == expected) {
-               _n = desired;
-               return true;
-           } else {
-               expected = _n;
-               return false;
-           }
-       }
-
-       bool compare_exchange_weak(int& expected, int desired, int mo = 5 /*memory_order_seq_cst*/)
-       {
-           return compare_exchange_strong(expected, desired, mo);
-       }
-   };
-
-   class fallback_atomic_bool {
-   private:
-       volatile bool _b;
-
-       fallback_atomic_bool(const fallback_atomic_bool&);
-       fallback_atomic_bool& operator=(const fallback_atomic_bool&);
-
-   public:
-       fallback_atomic_bool(bool b = false) : _b(b) {}
-
-       bool operator=(bool b)
-       {
-           _b = b;
-           return b;
-       }
-
-       operator bool() const
-       {
-           return _b;
-       }
-
-       bool load(int mo = 0 /*memory_order_relaxed*/) const
-       {
-           (void)mo;
-           return _b;
-       }
-
-       void store(bool b, int mo = 3 /*memory_order_release*/)
-       {
-           (void)mo;
-           _b = b;
-       }
-
-       bool exchange(bool val, int mo = 2 /*memory_order_acquire*/)
-       {
-           (void)mo;
-           bool old = _b;
-           _b = val;
-           return old;
-       }
-
-       bool compare_exchange_strong(bool& expected, bool desired, int mo = 5 /*memory_order_seq_cst*/)
-       {
-           (void)mo;
-           if (_b == expected) {
-               _b = desired;
-               return true;
-           } else {
-               expected = _b;
-               return false;
-           }
-       }
-   };
-
-   #define ATOMIC_INT fallback_atomic_int
-   #define ATOMIC_BOOL fallback_atomic_bool
 #else
-   #define ATOMIC_INT std::atomic_int
-   #define ATOMIC_BOOL std::atomic_bool
+
+    typedef int atomic_int_t;
+    #define LOAD_ATOMIC(a) (a)
+    #define STORE_ATOMIC(a, v)  ((a) = (v))
+    #define EXCHANGE_ATOMIC(a, v)  exchange_atomic_int((a), (v))
+    #define FETCH_ADD_ATOMIC(a, v)  fetch_add_atomic_int((a), (v))
+    #define COMPARE_EXCHANGE_ATOMIC(obj, expected, desired) \
+                                    compare_exchange_fallback((obj), (expected), (desired))
+
+    inline int exchange_atomic_int(int& a, int v)
+    {
+        int old = a;
+        a = v;
+        return old;
+    }
+
+    inline int fetch_add_atomic_int(int& a, int v)
+    {
+        int old = a;
+        a += v;
+        return old;
+    }
+
+    inline bool compare_exchange_fallback(int& obj, int& expected, int desired)
+    {
+        if (obj == expected) {
+            obj = desired;
+            return true;
+        } else {
+            expected = obj; // update expected on failure
+            return false;
+        }
+    }
 
 #endif
 

src/io/CompressedInputStream.cpp

@@ -76,8 +76,8 @@ CompressedInputStream::CompressedInputStream(InputStream& is,
     _available = 0;
     _entropyType = EntropyDecoderFactory::getType(entropy.c_str()); // throws on error
     _transformType = TransformFactory<byte>::getType(transform.c_str()); // throws on error
-    _initialized = false;
-    _closed = false;
+    _initialized = 0;
+    _closed = 0;
     _gcount = 0;
     _ibs = new DefaultInputBitStream(is, DEFAULT_BUFFER_SIZE);
     _jobs = tasks;
@@ -152,8 +152,8 @@ CompressedInputStream::CompressedInputStream(InputStream& is, Context& ctx, bool
     _available = 0;
     _entropyType = EntropyDecoderFactory::NONE_TYPE;
     _transformType = TransformFactory<byte>::NONE_TYPE;
-    _initialized = false;
-    _closed = false;
+    _initialized = 0;
+    _closed = 0;
     _gcount = 0;
     _ibs = new DefaultInputBitStream(is, DEFAULT_BUFFER_SIZE);
     _jobs = tasks;
@@ -326,15 +326,15 @@ void CompressedInputStream::submitBlock(int bufferId)
 int CompressedInputStream::_get(int inc)
 {
     try {
-        if (_initialized.load(memory_order_acquire) == false) {
+        if (LOAD_ATOMIC(_initialized) == 0) {
              readHeader();
 
              for (int i = 0; i < _jobs; i++)
                 submitBlock(i);
         }
 
         if (_available == 0) {
-            if (_closed.load(memory_order_relaxed) == true)
+            if (LOAD_ATOMIC(_closed) == 1)
                 throw ios_base::failure("Stream closed");
 
             DecodingTaskResult res;
@@ -427,7 +427,7 @@ istream& CompressedInputStream::read(char* data, streamsize length)
 
     while (remaining > 0) {
         // Reuse _get(0) logic logic implicitly
-        if (_initialized.load(memory_order_acquire) == false) {
+        if (LOAD_ATOMIC(_initialized) == 0) {
              readHeader();
 
              for (int i = 0; i < _jobs; i++)
@@ -507,7 +507,7 @@ istream& CompressedInputStream::read(char* data, streamsize length)
 
 void CompressedInputStream::readHeader()
 {
-    if (_initialized.exchange(true, memory_order_acquire))
+    if (EXCHANGE_ATOMIC(_initialized, 1) == 1)
         return;
 
     if (_headless == true)
@@ -695,11 +695,11 @@ bool CompressedInputStream::removeListener(Listener<Event>& bl)
 
 void CompressedInputStream::close()
 {
-    if (_closed.exchange(true, memory_order_relaxed))
+    if (EXCHANGE_ATOMIC(_closed, 1) == 1)
         return;
 
     // Signal to break the spin-loops in DecodingTask::run immediately.
-    _blockId.store(CANCEL_TASKS_ID, memory_order_release);
+    STORE_ATOMIC(_blockId, CANCEL_TASKS_ID);
 
 #ifdef CONCURRENCY_ENABLED
     // We must ensure no thread is writing to _buffers before we delete them.
@@ -748,7 +748,7 @@ void CompressedInputStream::notifyListeners(vector<Listener<Event>*>& listeners,
 template <class T>
 DecodingTask<T>::DecodingTask(SliceArray<byte>* iBuffer, SliceArray<byte>* oBuffer,
     int blockSize, DefaultInputBitStream* ibs, XXHash32* hasher32, XXHash64* hasher64,
-    ATOMIC_INT* processedBlockId, vector<Listener<Event>*>& listeners,
+    atomic_int_t* processedBlockId, vector<Listener<Event>*>& listeners,
     const Context& ctx)
     : _listeners(listeners)
     , _ctx(ctx)
@@ -779,7 +779,7 @@ T DecodingTask<T>::run()
 
     // Lock free synchronization
     while (true) {
-        const int taskId = _processedBlockId->load(memory_order_acquire);
+        const int taskId = LOAD_ATOMIC(*_processedBlockId);
 
         if (taskId == CompressedInputStream::CANCEL_TASKS_ID) {
             // Skip, an error occurred
@@ -807,12 +807,12 @@ T DecodingTask<T>::run()
         uint64 read = _ibs->readBits(lr);
 
         if (read == 0) {
-            _processedBlockId->store(CompressedInputStream::CANCEL_TASKS_ID, memory_order_release);
+            STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
             return T(*_data, blockId, 0, 0, 0, "Success");
         }
 
         if (read > (uint64(1) << 34)) {
-            _processedBlockId->store(CompressedInputStream::CANCEL_TASKS_ID, memory_order_release);
+            STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
             return T(*_data, blockId, 0, 0, Error::ERR_BLOCK_SIZE, "Invalid block size");
         }
 
@@ -835,7 +835,7 @@ T DecodingTask<T>::run()
 
         // After completion of the bitstream reading, increment the block id.
         // It unblocks the task processing the next block (if any)
-        _processedBlockId->store(blockId, memory_order_release);
+        STORE_ATOMIC(*_processedBlockId, blockId);
 
         const int from = _ctx.getInt("from", 1);
         const int to = _ctx.getInt("to", CompressedInputStream::MAX_BLOCK_ID);
@@ -876,7 +876,7 @@ T DecodingTask<T>::run()
 
         if ((preTransformLength <= 0) || (preTransformLength > maxTransformSize)) {
             // Error => cancel concurrent decoding tasks
-            _processedBlockId->store(CompressedInputStream::CANCEL_TASKS_ID, memory_order_release);
+            STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
             stringstream ss;
             ss << "Invalid compressed block length: " << preTransformLength;
 
@@ -933,7 +933,7 @@ T DecodingTask<T>::run()
         if (ed->decode(_buffer->_array, 0, preTransformLength) != preTransformLength) {
             // Error => cancel concurrent decoding tasks
             delete ed;
-            _processedBlockId->store(CompressedInputStream::CANCEL_TASKS_ID, memory_order_release);
+            STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
             return T(*_data, blockId, 0, checksum1, Error::ERR_PROCESS_BLOCK,
                 "Entropy decoding failed");
         }
@@ -998,8 +998,8 @@ T DecodingTask<T>::run()
     }
     catch (const exception& e) {
         // Make sure to unfreeze next block
-        if (_processedBlockId->load(memory_order_acquire) == blockId - 1)
-            _processedBlockId->store(blockId, memory_order_release);
+	int curBlockId = blockId - 1;
+	COMPARE_EXCHANGE_ATOMIC(*_processedBlockId, curBlockId, blockId);
 
         if (transform != nullptr)
             delete transform;