Merge pull request emscripten-core#4 from rstz/add-multithreading

Make PThreadFS threadsafe

Author: rstz

pthreadfs/examples/emscripten-tests/multiple_threads.cpp

@@ -0,0 +1,51 @@
+// basic file operations
+#include <iostream>
+#include <fstream>
+#include <string>
+#include <vector>
+#include <thread>
+#include <emscripten.h>
+#include "pthreadfs.h"
+
+
+void threadMain(int msg) {
+  size_t thread_id = std::hash<std::thread::id>{}(std::this_thread::get_id());
+  std::ofstream myfile;
+  myfile.open ("filesystemaccess/multi_threading_example", std::ios_base::app);
+  myfile << "Writing from thread " << msg << " Id: " << thread_id << "   ";
+  myfile.close();
+  EM_ASM({console.log(`Wrote on thread ${$0}`);}, thread_id);
+  return;
+}
+
+int main () {
+  emscripten_init_pthreadfs();
+  EM_ASM({console.log("Hello from main")});
+  std::remove("filesystemaccess/multi_threading_example"); 
+
+  constexpr int number_of_threads = 10;
+
+  std::cout << "Proof that stdout works fine.\n";
+
+  std::vector<std::thread> threads;
+
+  for (int i = 0; i< number_of_threads; i++) {
+    std::thread thread(threadMain, i);
+    threads.push_back(std::move(thread));
+  }
+
+  std::ofstream myfile;
+  myfile.open ("filesystemaccess/multi_threading_example");
+  myfile << "Writing the main thread.\n";
+  myfile.close();
+  
+  for (int i = 0; i< number_of_threads; i++) {
+    threads[i].join();
+  }
+
+  EM_ASM({
+    console.log('Remember to check that the contents of file multi_threading_example are correct.');
+  });
+
+  return 0;
+}

pthreadfs/pthreadfs.cpp

@@ -33,6 +33,9 @@ void SyncToAsync::shutdown() {
 }
 
 void SyncToAsync::doWork(std::function<void(SyncToAsync::Callback)> newWork) {
+  // Use the doWorkMutex to prevent more than one doWork being in flight at a
+  // time, so that this is usable from multiple threads safely.
+  std::lock_guard<std::mutex> doWorkLock(doWorkMutex);
   // Send the work over.
   {
     std::lock_guard<std::mutex> lock(mutex);
@@ -302,4 +305,4 @@ void emscripten_init_pthreadfs() {
     init_fsafs(&resumeWrapper_v);
   });
   return;
-}
+}

pthreadfs/pthreadfs.h

@@ -190,11 +190,16 @@ class SyncToAsync {
   // Run some work on thread. This is a synchronous call, but the thread can do
   // async work for us. To allow us to know when the async work finishes, the
   // worker is given a function to call at that time.
+  //
+  // It is safe to call this method from multiple threads, as it locks itself.
+  // That is, you can create an instance of this and call it from multiple
+  // threads freely.
   void doWork(std::function<void(Callback)> newWork);
 
 private:
   std::thread thread;
   std::mutex mutex;
+  std::mutex doWorkMutex;
   std::condition_variable condition;
   std::function<void(Callback)> work;
   bool readyToWork = false;
@@ -221,4 +226,4 @@ void resumeWrapper_l(long retVal);
 
 void resumeWrapper_wasi(__wasi_errno_t retVal);
 
-#endif  // PTHREADFS_H
+#endif  // PTHREADFS_H

pthreadfs/src/pthreadfs.cpp

@@ -33,6 +33,9 @@ void SyncToAsync::shutdown() {
 }
 
 void SyncToAsync::doWork(std::function<void(SyncToAsync::Callback)> newWork) {
+  // Use the doWorkMutex to prevent more than one doWork being in flight at a
+  // time, so that this is usable from multiple threads safely.
+  std::lock_guard<std::mutex> doWorkLock(doWorkMutex);
   // Send the work over.
   {
     std::lock_guard<std::mutex> lock(mutex);
@@ -302,4 +305,4 @@ void emscripten_init_pthreadfs() {
     init_fsafs(&resumeWrapper_v);
   });
   return;
-}
+}

pthreadfs/src/pthreadfs.h

@@ -190,11 +190,16 @@ class SyncToAsync {
   // Run some work on thread. This is a synchronous call, but the thread can do
   // async work for us. To allow us to know when the async work finishes, the
   // worker is given a function to call at that time.
+  //
+  // It is safe to call this method from multiple threads, as it locks itself.
+  // That is, you can create an instance of this and call it from multiple
+  // threads freely.
   void doWork(std::function<void(Callback)> newWork);
 
 private:
   std::thread thread;
   std::mutex mutex;
+  std::mutex doWorkMutex;
   std::condition_variable condition;
   std::function<void(Callback)> work;
   bool readyToWork = false;
@@ -221,4 +226,4 @@ void resumeWrapper_l(long retVal);
 
 void resumeWrapper_wasi(__wasi_errno_t retVal);
 
-#endif  // PTHREADFS_H
+#endif  // PTHREADFS_H