Fix stale workspace pointers in reused thread queues

common_thread.h

@@ -106,7 +106,15 @@ typedef struct blas_queue {
   blas_arg_t *args;
   void *range_m;
   void *range_n;
+
+  /*
+   * sa and sb are caller-owned inputs. worker_sb reports the sb workspace
+   * used by the threading backend for the most recent invocation.
+   * Keeping these roles separate prevents a reused queue from treating a
+   * released workspace as input to its next invocation.
+   */
   void *sa, *sb;
+  void *worker_sb;
 
   struct blas_queue *next;
 
@@ -183,6 +191,7 @@ static __inline void blas_queue_init(blas_queue_t *queue){
 
   queue -> sa    = NULL;
   queue -> sb    = NULL;
+  queue -> worker_sb = NULL;
   queue-> next  = NULL;
 }
 

cpp_thread_test/CMakeLists.txt

@@ -29,6 +29,10 @@ if (CPP_THREAD_SAFETY_TEST)
   add_executable(dgemm_thread_safety_mixed dgemm_thread_safety_mixed.cpp)
   target_link_libraries(dgemm_thread_safety_mixed ${CPP_THREAD_SAFETY_LIBS})
   add_test(NAME dgemm_thread_safety_mixed COMMAND ${CMAKE_CURRENT_BINARY_DIR}/dgemm_thread_safety_mixed ${CPP_THREAD_SAFETY_DGEMM_MIXED_ARGS})
+
+  if (USE_THREAD AND (USE_OPENMP OR (NOT WIN32 AND NOT CYGWIN)))
+    add_test(NAME dgemm_thread_safety_mixed_callback COMMAND ${CMAKE_CURRENT_BINARY_DIR}/dgemm_thread_safety_mixed ${CPP_THREAD_SAFETY_DGEMM_MIXED_ARGS} --callback)
+  endif()
 endif()
 
 

cpp_thread_test/Makefile

@@ -3,6 +3,22 @@ include $(TOPDIR)/Makefile.system
 
 all :: dgemv_tester dgemm_tester dgemm_mixed_tester
 
+CALLBACK_THREAD_TEST =
+ifeq ($(SMP),1)
+ifeq ($(USE_OPENMP),1)
+CALLBACK_THREAD_TEST = 1
+else
+ifneq ($(OSNAME),$(filter $(OSNAME),WINNT CYGWIN_NT))
+CALLBACK_THREAD_TEST = 1
+endif
+endif
+endif
+
+ifeq ($(CALLBACK_THREAD_TEST),1)
+.PHONY : dgemm_mixed_callback_tester
+all :: dgemm_mixed_callback_tester
+endif
+
 dgemv_tester :
 	$(CXX) $(COMMON_OPT) -Wall -Wextra -Wshadow -std=c++11 dgemv_thread_safety.cpp ../$(LIBNAME) $(EXTRALIB) $(FEXTRALIB) -o dgemv_tester
 	./dgemv_tester
@@ -15,5 +31,10 @@ dgemm_mixed_tester : dgemm_tester
 	$(CXX) $(COMMON_OPT) -Wall -Wextra -Wshadow -std=c++11 dgemm_thread_safety_mixed.cpp ../$(LIBNAME) $(EXTRALIB) $(FEXTRALIB) -o dgemm_mixed_tester
 	./dgemm_mixed_tester
 
+ifeq ($(CALLBACK_THREAD_TEST),1)
+dgemm_mixed_callback_tester : dgemm_mixed_tester
+	./dgemm_mixed_tester --callback
+endif
+
 clean ::
 	rm -f dgemv_tester dgemm_tester dgemm_mixed_tester

cpp_thread_test/dgemm_thread_safety_mixed.cpp

@@ -14,6 +14,21 @@
 #endif
 #include "cpp_thread_safety_common.h"
 
+std::atomic<uint32_t> callbackInvocations(0);
+
+void thread_callback(int sync, openblas_dojob_callback doJob, int numJobs,
+                     size_t jobDataElementSize, void* jobData, int doJobData){
+	(void)sync;
+	callbackInvocations.fetch_add(1, std::memory_order_relaxed);
+	std::vector<std::thread> workers;
+	workers.reserve(numJobs);
+	char* jobs = static_cast<char*>(jobData);
+	for(int i=0; i<numJobs; i++)
+		workers.emplace_back(doJob, i, jobs + i * jobDataElementSize, doJobData);
+	for(auto& worker : workers)
+		worker.join();
+}
+
 void compute_dgemm_pair(std::vector<double>& transA, std::vector<double>& noTransA, std::vector<double>& B, double* firstOutput, double* secondOutput, const blasint randomMatSize, const bool sameVariant){
 	cblas_dgemm(CblasRowMajor, CblasTrans, CblasNoTrans, randomMatSize, 2, 2, 1.0, &transA[0], randomMatSize, &B[0], 2, 0.0, firstOutput, 2);
 	if (sameVariant)
@@ -48,26 +63,31 @@ int main(int argc, char* argv[]){
 	uint32_t numTestRounds = 200;
 	uint32_t maxHwThreads = GetMaxHwThreads();
 	bool sameVariant = false;
+	bool useCallback = false;
 
 	if (maxHwThreads < numConcurrentThreads)
 		numConcurrentThreads = maxHwThreads;
 
-	if (argc != 1 && argc != 4 && argc != 5){
-		std::cout<<"ERROR: expected zero arguments, or: <M> <threads> <rounds> [sameVariant]"<<std::endl;
+	std::vector<std::string> positionalArgs;
+	for (int i = 1; i < argc; i++){
+		std::cout<<argv[i]<<std::endl;
+		if (std::string(argv[i]) == "--callback")
+			useCallback = true;
+		else
+			positionalArgs.push_back(argv[i]);
+	}
+
+	if (!positionalArgs.empty() && positionalArgs.size() != 3 && positionalArgs.size() != 4){
+		std::cout<<"ERROR: expected: [<M> <threads> <rounds> [sameVariant]] [--callback]"<<std::endl;
 		return 1;
 	}
 
-	if(argc == 4 || argc == 5){
-		std::vector<std::string> cliArgs;
-		for (int i = 1; i < argc; i++){
-			cliArgs.push_back(argv[i]);
-			std::cout<<argv[i]<<std::endl;
-		}
-		randomMatSize = std::stoul(cliArgs[0]);
-		numConcurrentThreads = std::stoul(cliArgs[1]);
-		numTestRounds = std::stoul(cliArgs[2]);
-		if (argc == 5)
-			sameVariant = std::stoul(cliArgs[3]) != 0;
+	if(!positionalArgs.empty()){
+		randomMatSize = std::stoul(positionalArgs[0]);
+		numConcurrentThreads = std::stoul(positionalArgs[1]);
+		numTestRounds = std::stoul(positionalArgs[2]);
+		if (positionalArgs.size() == 4)
+			sameVariant = std::stoul(positionalArgs[3]) != 0;
 	}
 
 	FailIfThreadsAreZero(numConcurrentThreads);
@@ -92,6 +112,8 @@ int main(int argc, char* argv[]){
 	std::cout<<"Number of testing rounds : "<<numTestRounds<<'\n';
 	std::cout<<"Second DGEMM uses "<<(sameVariant ? "the same transpose variant" : "a different transpose variant")<<'\n';
 	std::cout<<"OpenBLAS internal threads : "<<openblas_get_num_threads()<<'\n';
+	if (useCallback)
+		std::cout<<"Thread execution backend : caller callback\n";
 	std::cout<<"This test will need "<<(static_cast<uint64_t>(matrixElements) * 2 * 8 + static_cast<uint64_t>(outputElements) * (2 + 2 * numConcurrentThreads) * 8)/static_cast<double>(1024*1024)<<" MiB of RAM\n"<<std::endl;
 
 	std::cout<<"Filling matrices with deterministic values..."<<std::flush;
@@ -110,6 +132,9 @@ int main(int argc, char* argv[]){
 	compute_dgemm_pair(transA, noTransA, B, &referenceFirst[0], &referenceSecond[0], randomMatSize, sameVariant);
 	std::cout<<"done\n";
 
+	if (useCallback)
+		openblas_set_threads_callback_function(thread_callback);
+
 	std::cout<<"Testing mixed CBLAS DGEMM thread safety\n";
 	std::cout<<"Launching "<<numConcurrentThreads<<" worker threads..."<<std::flush;
 	for(uint32_t i=0; i<numConcurrentThreads; i++){
@@ -128,6 +153,15 @@ int main(int argc, char* argv[]){
 	}
 	std::cout<<"done\n";
 
+	if (useCallback) {
+		const uint32_t invocations = callbackInvocations.load();
+		std::cout<<"Thread callback invocations: "<<invocations<<std::endl;
+		if (invocations == 0) {
+			std::cout<<"Thread callback was not invoked!"<<std::endl;
+			return 1;
+		}
+	}
+
 	std::cout<<"Mixed DGEMM mismatches: "<<mismatches<<std::endl;
 	if (mismatches != 0) {
 		std::cout<<"Mixed CBLAS DGEMM thread safety test FAILED!"<<std::endl;

driver/level2/sbmv_thread.c

@@ -349,7 +349,7 @@ int CNAME(BLASLONG n, BLASLONG k, FLOAT *alpha, FLOAT *a, BLASLONG lda, FLOAT *x
 #else
 	    ONE, ZERO,
 #endif
-	    (FLOAT*)(queue[i].sb), 1, buffer, 1, NULL, 0);
+	    (FLOAT*)(queue[i].worker_sb), 1, buffer, 1, NULL, 0);
   }
 
   AXPYU_K(n, 0, 0,

driver/others/blas_server.c

@@ -837,6 +837,7 @@ if (openblas_threads_callback_) {
 #endif
 
   routine = (int (*)(blas_arg_t *, void *, void *, double *, double *, BLASLONG))queue -> routine;
+  queue->worker_sb = queue->sb;
 
   if (queue -> mode & BLAS_LEGACY) {
     legacy_exec(routine, queue -> mode, queue -> args, queue -> sb);
@@ -1140,9 +1141,10 @@ if (!(queue -> mode & BLAS_COMPLEX)){
         /* Other types in future */
     }
 }
-queue->sb=sb;
     }
 
+    queue->worker_sb = sb;
+
 #ifdef MONITOR
 main_status[cpu] = MAIN_RUNNING2;
 #endif

driver/others/blas_server_omp.c

@@ -329,7 +329,6 @@ static void exec_threads(int thread_num, blas_queue_t *queue, int buf_index){
 
     if (sa == NULL) {
       sa = (void *)((BLASLONG)buffer + GEMM_OFFSET_A);
-      queue->sa=sa;
     }
 
     if (sb == NULL) {
@@ -378,10 +377,11 @@ fprintf(stderr,"UNHANDLED COMPLEX\n");
           /* Other types in future */
 	  }
       }
-      queue->sb=sb;
     }
   }
 
+  queue->worker_sb = sb;
+
   if (queue -> mode & BLAS_LEGACY) {
     legacy_exec(queue -> routine, queue -> mode, queue -> args, sb);
   } else

driver/others/blas_server_win32.c

@@ -311,9 +311,10 @@ static DWORD WINAPI blas_thread_server(void *arg){
             /* Other types in future */
 	    }
 	}
-	queue->sb=sb;
       }
 
+      queue->worker_sb = sb;
+
 #ifdef MONITOR
       main_status[cpu] = MAIN_RUNNING2;
 #endif
@@ -483,6 +484,7 @@ int exec_blas(BLASLONG num, blas_queue_t *queue){
   if ((num > 1) && queue -> next) exec_blas_async(1, queue -> next);
 
   routine = queue -> routine;
+  queue->worker_sb = queue->sb;
 
   if (queue -> mode & BLAS_LEGACY) {
     legacy_exec(routine, queue -> mode, queue -> args, queue -> sb);