yum-archive/TaSTT-Whisper

High-performance GPGPU inference of OpenAI's Whisper automatic speech recognition (ASR) model

git clone https://git.yummers.dev/yum-archive/TaSTT-Whisper

KonstantinSource codes8c4603c

master
branchesmasterall refs →
3.1 KiB144 linesraw
1#include "stdafx.h"
2#include "parallelFor.h"
3
4namespace
5{
6	class alignas( 64 ) ParallelForContext
7	{
8		volatile long threadIndex;
9		volatile HRESULT status;
10
11		alignas( 64 ) void* const context;
12		const Whisper::pfnParallelForCallback pfn;
13
14		static void __stdcall callbackStatic( PTP_CALLBACK_INSTANCE Instance, PVOID pv, PTP_WORK Work );
15
16	public:
17
18		ParallelForContext( void* ctx, Whisper::pfnParallelForCallback pfn );
19
20		PTP_WORK createWork();
21
22		HRESULT getStatus() const;
23	};
24
25	ParallelForContext::ParallelForContext( void* ctx, Whisper::pfnParallelForCallback callback ) :
26		threadIndex( 1 ),
27		status( S_FALSE ),
28		context( ctx ),
29		pfn( callback )
30	{ }
31
32	PTP_WORK ParallelForContext::createWork()
33	{
34		return CreateThreadpoolWork( &callbackStatic, this, nullptr );
35	}
36
37	void __stdcall ParallelForContext::callbackStatic( PTP_CALLBACK_INSTANCE Instance, PVOID pv, PTP_WORK Work )
38	{
39		ParallelForContext& context = *(ParallelForContext*)pv;
40		int ith = InterlockedIncrement( &context.threadIndex );
41		ith--;
42		const HRESULT hr = context.pfn( ith, context.context );
43		if( SUCCEEDED( hr ) )
44			return;
45		InterlockedCompareExchange( &context.status, hr, S_FALSE );
46	}
47
48	HRESULT ParallelForContext::getStatus() const
49	{
50		const HRESULT hr = status;
51		if( SUCCEEDED( hr ) )
52			return S_OK;
53		return hr;
54	}
55}
56
57namespace Whisper
58{
59	HRESULT parallelFor( pfnParallelForCallback pfn, int threadsCount, void* ctx )
60	{
61		if( threadsCount < 1 )
62			return E_BOUNDS;
63		if( threadsCount == 1 )
64			return pfn( 0, ctx );
65
66		ParallelForContext context{ ctx, pfn };
67
68		PTP_WORK const pw = context.createWork();
69		if( nullptr == pw )
70			return getLastHr();
71
72		for( int i = 1; i < threadsCount; i++ )
73			SubmitThreadpoolWork( pw );
74
75		const HRESULT hr0 = pfn( 0, ctx );
76
77		WaitForThreadpoolWorkCallbacks( pw, FALSE );
78		CloseThreadpoolWork( pw );
79
80		if( FAILED( hr0 ) )
81			return hr0;
82		return context.getStatus();
83	}
84}
85
86using namespace Whisper;
87
88ThreadPoolWork::~ThreadPoolWork()
89{
90	if( nullptr != work )
91	{
92		CloseThreadpoolWork( work );
93		work = nullptr;
94	}
95}
96
97HRESULT ThreadPoolWork::create()
98{
99	if( nullptr == work )
100	{
101		work = CreateThreadpoolWork( &callbackStatic, this, nullptr );
102		if( nullptr != work )
103			return S_OK;
104		return getLastHr();
105	}
106	return HRESULT_FROM_WIN32( ERROR_ALREADY_INITIALIZED );
107}
108
109HRESULT ThreadPoolWork::parallelFor( int threadsCount ) noexcept
110{
111	if( nullptr != work )
112	{
113		if( threadsCount <= 1 )
114			return threadPoolCallback( 0 );
115
116		threadIndex = 1;
117		status = S_FALSE;
118		for( int i = 1; i < threadsCount; i++ )
119			SubmitThreadpoolWork( work );
120
121		const HRESULT hr0 = threadPoolCallback( 0 );
122
123		WaitForThreadpoolWorkCallbacks( work, FALSE );
124
125		if( FAILED( hr0 ) )
126			return hr0;
127		if( SUCCEEDED( status ) )
128			return S_OK;
129		return status;
130	}
131
132	return OLE_E_BLANK;
133}
134
135void __stdcall ThreadPoolWork::callbackStatic( PTP_CALLBACK_INSTANCE Instance, PVOID pv, PTP_WORK Work )
136{
137	ThreadPoolWork* tpw = (ThreadPoolWork*)pv;
138	int ith = InterlockedIncrement( &tpw->threadIndex );
139	ith--;
140	const HRESULT hr = tpw->threadPoolCallback( ith );
141	if( SUCCEEDED( hr ) )
142		return;
143	InterlockedCompareExchange( &tpw->status, hr, S_FALSE );
144}