Subzero: Initial implementation of multithreaded translation.

src/IceGlobalContext.h

 //===- subzero/src/IceGlobalContext.h - Global context defs -----*- C++ -*-===//
 //
 //                        The Subzero Code Generator
 //
 // This file is distributed under the University of Illinois Open Source
 // License. See LICENSE.TXT for details.
 //
 //===----------------------------------------------------------------------===//
 //
 // This file declares aspects of the compilation that persist across
 // multiple functions.
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef SUBZERO_SRC_ICEGLOBALCONTEXT_H
 #define SUBZERO_SRC_ICEGLOBALCONTEXT_H
 
-#include <memory>
 #include <mutex>
+#include <queue>
+#include <thread>
 
 #include "IceDefs.h"
 #include "IceClFlags.h"
 #include "IceIntrinsics.h"
 #include "IceRNG.h"
 #include "IceTimerTree.h"
 #include "IceTypes.h"
 
 namespace Ice {
 
@@ skipping 59 matching lines @@
   class ThreadContext {
     ThreadContext(const ThreadContext &) = delete;
     ThreadContext &operator=(const ThreadContext &) = delete;
 
   public:
     ThreadContext() {}
     CodeStats StatsFunction;
     std::vector<TimerStack> Timers;
   };
 
+  // CfgQueue is the translation work queue.  It allows multiple
+  // producers and multiple consumers (though currently only a single
+  // producer is used).  The producer adds entries using add(), and
+  // may block if the queue is "full" to control Cfg memory footprint.
+  // The producer uses end() to indicate that no more entries will be
+  // added.  The consumer removes an item using get(), which will
+  // return nullptr if end() has been called and the queue is empty.
+  //
+  // The MaxSize ctor arg controls the maximum size the queue can grow
+  // to.  The Sequential arg indicates purely sequential execution in
+  // which the single thread should never wait().
+  //
+  // Two condition variables are used in the implementation.
+  // GrewOrEnded signals waiting workers that the producer has changed
+  // the state of the queue.  Shrunk signals a blocked producer that a
+  // consumer has changed the state of the queue.
+  class CfgQueue {
+  public:
+    CfgQueue(size_t MaxSize, bool Sequential)
+        : MaxSize(MaxSize), Sequential(Sequential), IsEnded(false) {}
+    void add(Cfg *Func) {
+      std::unique_lock<GlobalLockType> L(Lock);
+      // If the work queue is already "full", wait for a consumer to
+      // grab an element and shrink the queue.
+      while (!Sequential && WorkQueue.size() >= MaxSize) {
+        Shrunk.wait(L);
+      }
+      WorkQueue.push(Func);
+      L.unlock();
+      GrewOrEnded.notify_one();
+    }
+    Cfg *get() {
+      std::unique_lock<GlobalLockType> L(Lock);
+      while (!IsEnded || !WorkQueue.empty()) {
+        if (!WorkQueue.empty()) {
+          Cfg *Func = WorkQueue.front();
+          WorkQueue.pop();
+          L.unlock();
+          Shrunk.notify_one();
+          return Func;
+        }
+        // If the work queue is empty, and this is pure sequential
+        // execution, then return nullptr.
+        if (Sequential)
+          return nullptr;
+        GrewOrEnded.wait(L);
+      }
+      return nullptr;
+    }
+    void end() {
+      std::unique_lock<GlobalLockType> L(Lock);
+      IsEnded = true;
+      L.unlock();
+      GrewOrEnded.notify_all();
+    }
+
+  private:
+    std::queue<Cfg *> WorkQueue;
+    // Lock guards access to WorkQueue and IsEnded.
+    GlobalLockType Lock;
+    // GrewOrEnded is notified (by the producer) when something is
+    // added to the queue, in case consumers are waiting for a
+    // non-empty queue.
+    std::condition_variable GrewOrEnded;
+    // Shrunk is notified (by the consumer) when something is removed
+    // from the queue, in case the producer is waiting for the queue
+    // to drop below maximum capacity.
+    std::condition_variable Shrunk;
+    const size_t MaxSize;
+    const bool Sequential;
+    bool IsEnded;

[JF, 2015/01/23 17:22:43]

I think it's worth reordering and aligning to cacheline the members so they're
grouped as they're accessed most of the time (I'm thinking MOESI):
 - Written only by the producer but read by consumers (GrewOrEnded).
 - Written only by the consumers but read by producer (Shrunk).
 - Only be read by all (MaxSize/Sequential/IsEnded).
 - Read/write by all (Lock/WorkQueue).

This calls out how WorkQueue is used: it's probably better as a circular buffer,
where the front and back pointers are read/written exclusively by the
producer/consumer so they would be better positioned in different cachelines.
The buffer itself may end up aliasing when there's less work or when it gets too
full, but that's OK. I would really use a statically sized array for this, sized
to a power of 2, and use front/back with mod-of-a-constant-power-of-2 to find
elements.

Reordering as I suggest would make it perform better, but it also makes it way
easier to understand the design if the access patterns are explained properly.

[Jim Stichnoth, 2015/01/23 21:54:02]

Done.  And added a TODO about considering std::array for a circular buffer.

+  };
+
 public:
   GlobalContext(Ostream *OsDump, Ostream *OsEmit, ELFStreamer *ELFStreamer,
                 VerboseMask Mask, TargetArch Arch, OptLevel Opt,
                 IceString TestPrefix, const ClFlags &Flags);
   ~GlobalContext();
 
-  // Returns true if any of the specified options in the verbose mask
-  // are set.  If the argument is omitted, it checks if any verbose
-  // options at all are set.
   VerboseMask getVerbose() const { return VMask; }
-  bool isVerbose(VerboseMask Mask = IceV_All) const { return VMask & Mask; }
-  void setVerbose(VerboseMask Mask) { VMask = Mask; }
-  void addVerbose(VerboseMask Mask) { VMask |= Mask; }
-  void subVerbose(VerboseMask Mask) { VMask &= ~Mask; }
 
   // The dump and emit streams need to be used by only one thread at a
   // time.  This is done by exclusively reserving the streams via
   // lockStr() and unlockStr().  The OstreamLocker class can be used
   // to conveniently manage this.
   //
   // The model is that a thread grabs the stream lock, then does an
   // arbitrary amount of work during which far-away callees may grab
   // the stream and do something with it, and finally the thread
   // releases the stream lock.  This allows large chunks of output to
   // be dumped or emitted without risking interleaving from multiple
   // threads.
   void lockStr() { StrLock.lock(); }
   void unlockStr() { StrLock.unlock(); }
   Ostream &getStrDump() { return *StrDump; }
   Ostream &getStrEmit() { return *StrEmit; }
 
   TargetArch getTargetArch() const { return Arch; }
   OptLevel getOptLevel() const { return Opt; }
+  std::error_code getErrorStatus() const { return ErrorStatus; }
 
   // When emitting assembly, we allow a string to be prepended to
   // names of translated functions.  This makes it easier to create an
   // execution test against a reference translator like llc, with both
   // translators using the same bitcode as input.
   IceString getTestPrefix() const { return TestPrefix; }
   IceString mangleName(const IceString &Name) const;
 
   // Manage Constants.
   // getConstant*() functions are not const because they might add
@@ skipping 80 matching lines @@
 
   TimerStackIdT newTimerStackID(const IceString &Name);
   TimerIdT getTimerID(TimerStackIdT StackID, const IceString &Name);
   void pushTimer(TimerIdT ID, TimerStackIdT StackID = TSK_Default);
   void popTimer(TimerIdT ID, TimerStackIdT StackID = TSK_Default);
   void resetTimer(TimerStackIdT StackID);
   void setTimerName(TimerStackIdT StackID, const IceString &NewName);
   void dumpTimers(TimerStackIdT StackID = TSK_Default,
                   bool DumpCumulative = true);
 
+  // Adds a newly parsed and constructed function to the Cfg work
+  // queue.  Notifies any idle workers that a new function is
+  // available for translating.  May block if the work queue is too
+  // large, in order to control memory footprint.
+  void cfgQueueAdd(Cfg *Func) { CfgQ.add(Func); }
+  // Takes a Cfg from the work queue for translating.  May block if
+  // the work queue is currently empty.  Returns nullptr if there is
+  // no more work - the queue is empty and either end() has been
+  // called or the Sequential flag was set.
+  Cfg *cfgQueueGet() { return CfgQ.get(); }
+  // Notifies that no more work will be added to the work queue.
+  void cfgQueueEnd() { CfgQ.end(); }
+
+  void startWorkerThreads() {
+    size_t NumWorkers = getFlags().NumTranslationThreads;
+    for (size_t i = 0; i < NumWorkers; ++i) {
+      ThreadContext *WorkerTLS = new ThreadContext();
+      AllThreadContexts.push_back(WorkerTLS);
+      TranslationThreads.push_back(std::thread(
+          &GlobalContext::translateFunctionsWrapper, this, WorkerTLS));
+    }
+    if (NumWorkers) {
+      // TODO(stichnot): start a new thread for the emitter queue worker.
+    }
+  }
+
+  void waitForWorkerThreads() {
+    cfgQueueEnd();
+    // TODO(stichnot): call end() on the emitter work queue.
+    for (std::thread &Worker : TranslationThreads) {
+      Worker.join();
+    }
+    TranslationThreads.clear();
+    // TODO(stichnot): join the emitter thread.
+  }
+
+  // Translation thread startup routine.
+  void translateFunctionsWrapper(ThreadContext *MyTLS) {
+    TLS = MyTLS;
+    translateFunctions();
+  }
+  // Translate functions from the Cfg queue until the queue is empty.
+  void translateFunctions();
+
+  // Utility function to match a symbol name against a match string.
+  // This is used in a few cases where we want to take some action on
+  // a particular function or symbol based on a command-line argument,
+  // such as changing the verbose level for a particular function.  An
+  // empty Match argument means match everything.  Returns true if
+  // there is a match.
+  static bool matchSymbolName(const IceString &SymbolName,
+                              const IceString &Match) {
+    return Match.empty() || Match == SymbolName;
+  }
+
 private:
   // Try to make sure the mutexes are allocated on separate cache
   // lines, assuming the maximum cache line size is 64.
   const static size_t MaxCacheLineSize = 64;
   alignas(MaxCacheLineSize) GlobalLockType AllocLock;
   alignas(MaxCacheLineSize) GlobalLockType ConstPoolLock;
   alignas(MaxCacheLineSize) GlobalLockType StatsLock;
   alignas(MaxCacheLineSize) GlobalLockType TimerLock;
 
   // StrLock is a global lock on the dump and emit output streams.
   typedef std::mutex StrLockType;
   StrLockType StrLock;
 
   Ostream *StrDump; // Stream for dumping / diagnostics
   Ostream *StrEmit; // Stream for code emission
 
   ArenaAllocator<> Allocator;
   VerboseMask VMask;
   std::unique_ptr<ConstantPool> ConstPool;
   Intrinsics IntrinsicsInfo;
   const TargetArch Arch;
   const OptLevel Opt;
   const IceString TestPrefix;
   const ClFlags &Flags;
   RandomNumberGenerator RNG;
   std::unique_ptr<ELFObjectWriter> ObjectWriter;
   CodeStats StatsCumulative;
   std::vector<TimerStack> Timers;
+  CfgQueue CfgQ;
+  std::error_code ErrorStatus;
 
   LockedPtr<ArenaAllocator<>> getAllocator() {
     return LockedPtr<ArenaAllocator<>>(&Allocator, &AllocLock);
   }
   LockedPtr<ConstantPool> getConstPool() {
     return LockedPtr<ConstantPool>(ConstPool.get(), &ConstPoolLock);
   }
   LockedPtr<CodeStats> getStatsCumulative() {
     return LockedPtr<CodeStats>(&StatsCumulative, &StatsLock);
   }
   LockedPtr<std::vector<TimerStack>> getTimers() {
     return LockedPtr<std::vector<TimerStack>>(&Timers, &TimerLock);
   }
 
   std::vector<ThreadContext *> AllThreadContexts;
+  std::vector<std::thread> TranslationThreads;
   // Each thread has its own TLS pointer which is also held in
   // AllThreadContexts.
   thread_local static ThreadContext *TLS;
 
   // Private helpers for mangleName()
   typedef llvm::SmallVector<char, 32> ManglerVector;
   void incrementSubstitutions(ManglerVector &OldName) const;
 };
 
 // Helper class to push and pop a timer marker.  The constructor
@@ skipping 37 matching lines @@
   explicit OstreamLocker(GlobalContext *Ctx) : Ctx(Ctx) { Ctx->lockStr(); }
   ~OstreamLocker() { Ctx->unlockStr(); }
 
 private:
   GlobalContext *const Ctx;
 };
 
 } // end of namespace Ice
 
 #endif // SUBZERO_SRC_ICEGLOBALCONTEXT_H