Implemented async decode (multiple threads still buggy)

QuintetPlayer.pro

@@ -28,13 +28,16 @@ SOURCES += \
         src/mainwindow.cpp \
     src/clHCA.cpp \
     src/utils.cpp \
-    src/HCAStreamChannel.cpp
+    src/HCAStreamChannel.cpp \
+    src/HCADecodeService.cpp
 
 HEADERS += \
         src/mainwindow.h \
     src/clHCA.h \
     src/utils.h \
     src/HCAStreamChannel.h \
+    src/HCADecodeService.h \
+    src/Semaphore.h \
     bass/bass.h \
     bass/bassmix.h
 

src/HCADecodeService.cpp

@@ -0,0 +1,145 @@
+#include <map>
+#include "bass.h"
+#include "HCADecodeService.h"
+#include "clHCA.h"
+
+// Currently multithreaded decode is bugged, please use single thread for now
+HCADecodeService::HCADecodeService(unsigned int numthreads)
+    : mainsem(numthreads),
+      datasem{ 0 },
+      numchannels{0},
+      workingrequest{nullptr},
+      channels{nullptr},
+      shutdown{false}
+{
+    if(numthreads > 0)
+    {
+        this->numthreads = numthreads;
+    }
+    else
+    {
+        this->numthreads = 1;
+    }
+    workingblocks = new int[numthreads];
+	for (unsigned int i = 0; i < numthreads; ++i)
+	{
+		workersem.emplace_back(0);
+		worker_threads.emplace_back(&HCADecodeService::Decode_Thread, this, i);
+		workingblocks[i] = -1;
+	}
+	dispatchthread = std::thread{ &HCADecodeService::Main_Thread, this };
+}
+
+HCADecodeService::~HCADecodeService()
+{
+	shutdown = true;
+	datasem.notify();
+	dispatchthread.join();
+}
+
+void HCADecodeService::cancel_decode(void* ptr)
+{
+	mutex.lock();
+	for (auto it = filelist.begin(); it != filelist.end(); ++it)
+	{
+        if (it->first.first == ptr)
+		{
+			filelist.erase(it);
+			break;
+		}
+	}
+	if (workingrequest == ptr)
+	{
+		blocks.clear();
+    }
+	mutex.unlock();
+    for (unsigned int i = 0; i < numthreads; ++i)
+    {
+        while (workingblocks[i] != -1); // busy wait until threads are finished
+    }
+}
+
+std::pair<void*, size_t> HCADecodeService::decode(const std::string& filename, DWORD samplenum)
+{
+	clHCA hca(0xBC731A85, 0x0002B875);
+	void* wavptr = nullptr;
+	size_t sz = 0;
+	hca.Analyze(wavptr, sz, filename.c_str());
+	if (wavptr != nullptr)
+	{
+        unsigned int blocknum = samplenum/(1024 * hca.get_channelCount());
+		mutex.lock();
+        filelist[std::make_pair(wavptr, blocknum)] = std::move(hca);
+		mutex.unlock();
+		datasem.notify();
+	}
+	return std::pair<void*, size_t>(wavptr, sz);
+}
+
+void HCADecodeService::Main_Thread()
+{
+	while (true)
+	{
+		datasem.wait();
+		if (shutdown)
+		{
+			break;
+		}
+		mutex.lock();
+		auto it = filelist.begin();
+        workingrequest = it->first.first;
+		workingfile = std::move(it->second);
+        unsigned blocknum = it->first.second;
+		filelist.erase(it);
+		numchannels = workingfile.get_channelCount();
+		channels = new clHCA::stChannel[numchannels * numthreads];
+		workingfile.PrepDecode(channels, numthreads);
+		unsigned int blockCount = workingfile.get_blockCount();
+        for (unsigned int i = blocknum, j = 0; j < blockCount; ++i, ++j)
+		{
+            blocks.push_back(i % blockCount);
+		}
+		while (!blocks.empty())
+		{
+			mutex.unlock();
+			mainsem.wait();
+            for (unsigned int i = 0; i < numthreads; ++i)
+			{
+				mutex.lock();
+				if (workingblocks[i] == -1 && !blocks.empty())
+				{
+					workingblocks[i] = blocks.front();
+					blocks.pop_front();
+					workersem[i].notify();
+				}
+				mutex.unlock();
+			}
+			mainsem.notify();
+			mutex.lock();
+		}
+		mutex.unlock();
+        for (unsigned int i = 0; i < numthreads; ++i)
+		{
+			while (workingblocks[i] != -1); // busy wait until threads are finished
+		}
+		delete[] channels;
+	}
+	for (int i = 0; i < numthreads; ++i)
+	{
+		workersem[i].notify();
+		worker_threads[i].join();
+	}
+}
+
+void HCADecodeService::Decode_Thread(int id)
+{
+	workersem[id].wait();
+	while (workingblocks[id] != -1)
+	{
+		mainsem.wait();
+		workingfile.AsyncDecode(channels + (id * numchannels), workingblocks[id], (unsigned char*)workingrequest);
+		workingblocks[id] = -1;
+		mainsem.notify();
+		workersem[id].wait();
+	}
+}

src/HCADecodeService.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <thread>
+#include <deque>
+#include <map>
+#include <string>
+#include "Semaphore.h"
+#include "clHCA.h"
+
+class HCADecodeService
+{
+public:
+	HCADecodeService(unsigned int num_threads);
+	~HCADecodeService();
+	void cancel_decode(void* ptr);
+    std::pair<void*, size_t> decode(const std::string& hcafilename, DWORD samplenum);
+	//void wait_for_finish();
+private:
+	void Decode_Thread(int id);
+	void Main_Thread();
+	clHCA workingfile;
+	unsigned int numthreads, requestnum, numchannels;
+	void* workingrequest;
+	std::thread dispatchthread;
+	std::deque<std::thread> worker_threads;
+    std::map<std::pair<void*, unsigned int>, clHCA> filelist;
+	std::deque<unsigned int> blocks;
+	int* workingblocks;
+	std::deque<Semaphore> workersem;
+	Semaphore mainsem, datasem;
+	std::mutex mutex;
+	clHCA::stChannel* channels;
+	bool shutdown;
+};
+

src/HCAStreamChannel.cpp

@@ -3,8 +3,9 @@
 
 
 
-HCAStreamChannel::HCAStreamChannel()
+HCAStreamChannel::HCAStreamChannel(HCADecodeService* dec)
 {
+	this->dec = dec;
 	this->flags = 0;
 	ptr = nullptr;
 	size = 0;
@@ -16,7 +17,7 @@ HCAStreamChannel::HCAStreamChannel(const HCAStreamChannel& other)
 {
 	if (other.ptr == nullptr)
 	{
-		HCAStreamChannel();
+		HCAStreamChannel(nullptr);
 		return;
 	}
 	size = other.size;
@@ -25,10 +26,11 @@ HCAStreamChannel::HCAStreamChannel(const HCAStreamChannel& other)
 	__load();
 }
 
-HCAStreamChannel::HCAStreamChannel(const std::string& filename)
+HCAStreamChannel::HCAStreamChannel(HCADecodeService* dec, const std::string& filename)
 {
+	this->dec = dec;
 	this->flags = 0;
-	load(filename);
+    load(filename, 0);
 }
 
 
@@ -37,39 +39,78 @@ HCAStreamChannel::~HCAStreamChannel()
 	unload();
 }
 
+HCAStreamChannel& HCAStreamChannel::operator=(HCAStreamChannel&& other)
+{
+    unload();
+    if (this != &other)
+    {
+        ptr = other.ptr;
+        size = other.size;
+        playback_channel = other.playback_channel;
+        decode_channel = other.decode_channel;
+        flags = other.flags;
+        other.ptr = nullptr;
+        other.size = 0;
+        other.playback_channel = 0;
+        other.decode_channel = 0;
+    }
+    return *this;
+}
+
 void HCAStreamChannel::unload()
 {
-    destroy_channels();
+	dec->cancel_decode(ptr);
+	if (playback_channel != 0) { BASS_ChannelStop(playback_channel); playback_channel = 0; }
+	if (decode_channel != 0) { BASS_ChannelStop(decode_channel); decode_channel = 0; }
 	if (ptr != nullptr) { delete[] ptr; ptr = nullptr; }
 	size = 0;
 }
 
 bool HCAStreamChannel::load(const std::string& filename)
 {
-	clHCA hca(0xBC731A85, 0x0002B875);
-	ptr = hca.DecodeToMemory(size, filename.c_str());
+    auto pair = dec->decode(filename, 0);
+	ptr = pair.first;
+	size = pair.second;
 	return __load();
 }
 
+bool HCAStreamChannel::load(const std::string& filename, DWORD samplenum)
+{
+    auto pair = dec->decode(filename, samplenum);
+    ptr = pair.first;
+    size = pair.second;
+    return __load();
+}
+
 bool HCAStreamChannel::__load()
 {
 	if (ptr == nullptr)
 	{
 		playback_channel = 0;
 		decode_channel = 0;
 		return false;
-    }
-    make_channels();
-    if (playback_channel == 0 || decode_channel == 0)
+	}
+	playback_channel = BASS_StreamCreateFile(TRUE, ptr, 0, size, flags);
+	if (playback_channel == 0)
 	{
-        destroy_channels();
+        dec->cancel_decode(ptr);
 		delete[] ptr;
 		ptr = nullptr;
-        playback_channel = 0;
 		decode_channel = 0;
 		size = 0;
 		return false;
-    }
+	}
+	decode_channel = BASS_StreamCreateFile(TRUE, ptr, 0, size, flags | BASS_STREAM_DECODE);
+	if (decode_channel == 0)
+	{
+        dec->cancel_decode(ptr);
+		delete[] ptr;
+		ptr = nullptr;
+		BASS_ChannelStop(playback_channel);
+		playback_channel = 0;
+		size = 0;
+		return false;
+	}
 	return true;
 }
 

src/HCAStreamChannel.h

@@ -2,33 +2,19 @@
 
 #include <string>
 #include "bass.h"
+#include "HCADecodeService.h"
 
 class HCAStreamChannel
 {
 public:
-	HCAStreamChannel();
-    HCAStreamChannel(const HCAStreamChannel& other);
-    HCAStreamChannel& operator=(HCAStreamChannel&& other)
-    {
-        unload();
-        if (this != &other)
-        {
-            ptr = other.ptr;
-            size = other.size;
-            playback_channel = other.playback_channel;
-            decode_channel = other.decode_channel;
-            flags = other.flags;
-            other.ptr = nullptr;
-            other.size = 0;
-            other.playback_channel = 0;
-            other.decode_channel = 0;
-        }
-        return *this;
-    }
-	HCAStreamChannel(const std::string& filename);
+	HCAStreamChannel(HCADecodeService* dec);
+	HCAStreamChannel(const HCAStreamChannel& other);
+    HCAStreamChannel(HCADecodeService* dec, const std::string& filename);
+    HCAStreamChannel& operator=(HCAStreamChannel&& other);
 	~HCAStreamChannel();
 	void unload();
-	bool load(const std::string& filename);
+    bool load(const std::string& filename);
+    bool load(const std::string& filename, DWORD samplenum);
 	bool valid();
 	DWORD get_playback_channel();
 	DWORD get_decode_channel();
@@ -37,6 +23,7 @@ class HCAStreamChannel
     void make_channels();
 private:
 	bool __load();
+	HCADecodeService* dec;
 	void* ptr;
 	size_t size;
 	DWORD playback_channel, decode_channel;