summaryrefslogtreecommitdiffstatshomepage
path: root/src/osd/modules/sync/work_osd.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/osd/modules/sync/work_osd.cpp')
-rw-r--r--src/osd/modules/sync/work_osd.cpp297
1 files changed, 143 insertions, 154 deletions
diff --git a/src/osd/modules/sync/work_osd.cpp b/src/osd/modules/sync/work_osd.cpp
index 9f80c9aec76..3afe4e6c273 100644
--- a/src/osd/modules/sync/work_osd.cpp
+++ b/src/osd/modules/sync/work_osd.cpp
@@ -20,6 +20,7 @@
#include <mutex>
#include <atomic>
#include <thread>
+#include <vector>
// MAME headers
#include "osdcore.h"
@@ -29,15 +30,9 @@
#include "eminline.h"
-#if defined(SDLMAME_MACOSX)
-#include "osxutils.h"
-#endif
#if defined(SDLMAME_LINUX) || defined(SDLMAME_BSD) || defined(SDLMAME_HAIKU) || defined(SDLMAME_EMSCRIPTEN) || defined(SDLMAME_MACOSX)
#include <pthread.h>
#endif
-#if defined(OSD_SDL)
-typedef void *PVOID;
-#endif
//============================================================
// DEBUGGING
@@ -96,10 +91,26 @@ static void spin_while_not(const volatile _AtomType * volatile atom, const _Main
struct work_thread_info
{
- osd_work_queue * queue; // pointer back to the queue
+ work_thread_info(UINT32 aid, osd_work_queue &aqueue)
+ : queue(aqueue)
+ , handle(nullptr)
+ , wakeevent(FALSE, FALSE) // auto-reset, not signalled
+ , active(0)
+ , id(aid)
+#if KEEP_STATISTICS
+ , itemsdone(0)
+ , actruntime(0)
+ , runtime(0)
+ , spintime(0)
+ , waittime(0)
+#endif
+ {
+ }
+ osd_work_queue & queue; // pointer back to the queue
std::thread * handle; // handle to the thread
- osd_event * wakeevent; // wake event for the thread
+ osd_event wakeevent; // wake event for the thread
std::atomic<INT32> active; // are we actively processing work?
+ UINT32 id;
#if KEEP_STATISTICS
INT32 itemsdone;
@@ -113,7 +124,28 @@ struct work_thread_info
struct osd_work_queue
{
- std::mutex *lock; // lock for protecting the queue
+ osd_work_queue()
+ : list(nullptr)
+ , tailptr(nullptr)
+ , free(nullptr)
+ , items(0)
+ , livethreads(0)
+ , waiting(0)
+ , exiting(0)
+ , threads(0)
+ , flags(0)
+ , doneevent(TRUE, TRUE) // manual reset, signalled
+#if KEEP_STATISTICS
+ , itemsqueued(0)
+ , setevents(0)
+ , extraitems(0)
+ , spinloops(0)
+#endif
+ {
+
+ }
+
+ std::mutex lock; // lock for protecting the queue
std::atomic<osd_work_item *> list; // list of items in the queue
osd_work_item ** volatile tailptr; // pointer to the tail pointer of work items in the queue
std::atomic<osd_work_item *> free; // free list of work items
@@ -123,8 +155,8 @@ struct osd_work_queue
std::atomic<INT32> exiting; // should the threads exit on their next opportunity?
UINT32 threads; // number of threads in this queue
UINT32 flags; // creation flags
- work_thread_info * thread; // array of thread information
- osd_event * doneevent; // event signalled when work is complete
+ std::vector<work_thread_info *> thread; // array of thread information
+ osd_event doneevent; // event signalled when work is complete
#if KEEP_STATISTICS
std::atomic<INT32> itemsqueued; // total items queued
@@ -137,8 +169,20 @@ struct osd_work_queue
struct osd_work_item
{
+ osd_work_item(osd_work_queue &aqueue)
+ : next(nullptr)
+ , queue(aqueue)
+ , callback(nullptr)
+ , param(nullptr)
+ , result(nullptr)
+ , event(nullptr) // manual reset, not signalled
+ , flags(0)
+ , done(FALSE)
+ {
+ }
+
osd_work_item * next; // pointer to next item
- osd_work_queue * queue; // pointer back to the owning queue
+ osd_work_queue & queue; // pointer back to the owning queue
osd_work_callback callback; // callback function
void * param; // callback parameter
void * result; // callback result
@@ -204,25 +248,12 @@ osd_work_queue *osd_work_queue_alloc(int flags)
const char *osdworkqueuemaxthreads = osd_getenv(ENV_WORKQUEUEMAXTHREADS);
// allocate a new queue
- queue = (osd_work_queue *)osd_malloc(sizeof(*queue));
- if (queue == NULL)
- goto error;
- memset(queue, 0, sizeof(*queue));
+ queue = new osd_work_queue();
// initialize basic queue members
queue->tailptr = (osd_work_item **)&queue->list;
queue->flags = flags;
- // allocate events for the queue
- queue->doneevent = osd_event_alloc(TRUE, TRUE); // manual reset, signalled
- if (queue->doneevent == NULL)
- goto error;
-
- // initialize the critical section
- queue->lock = new std::mutex();
- if (queue->lock == NULL)
- goto error;
-
// determine how many threads to create...
// on a single-CPU system, create 1 thread for I/O queues, and 0 threads for everything else
if (numprocs == 1)
@@ -247,23 +278,13 @@ osd_work_queue *osd_work_queue_alloc(int flags)
printf("osdprocs: %d effecprocs: %d threads: %d allocthreads: %d osdthreads: %d maxthreads: %d queuethreads: %d\n", osd_num_processors, numprocs, threadnum, allocthreadnum, osdthreadnum, WORK_MAX_THREADS, queue->threads);
#endif
- queue->thread = (work_thread_info *)osd_malloc_array(allocthreadnum * sizeof(queue->thread[0]));
- if (queue->thread == NULL)
- goto error;
- memset(queue->thread, 0, allocthreadnum * sizeof(queue->thread[0]));
+ for (threadnum = 0; threadnum < allocthreadnum; threadnum++)
+ queue->thread.push_back(new work_thread_info(threadnum, *queue));
// iterate over threads
for (threadnum = 0; threadnum < queue->threads; threadnum++)
{
- work_thread_info *thread = &queue->thread[threadnum];
-
- // set a pointer back to the queue
- thread->queue = queue;
-
- // create the per-thread wake event
- thread->wakeevent = osd_event_alloc(FALSE, FALSE); // auto-reset, not signalled
- if (thread->wakeevent == NULL)
- goto error;
+ work_thread_info *thread = queue->thread[threadnum];
// create the thread
thread->handle = new std::thread(worker_thread_entry, thread);
@@ -281,7 +302,7 @@ osd_work_queue *osd_work_queue_alloc(int flags)
// start a timer going for "waittime" on the main thread
if (flags & WORK_QUEUE_FLAG_MULTI)
{
- begin_timing(queue->thread[queue->threads].waittime);
+ begin_timing(queue->thread[queue->threads]->waittime);
}
return queue;
@@ -319,7 +340,7 @@ int osd_work_queue_wait(osd_work_queue *queue, osd_ticks_t timeout)
// if this is a multi queue, help out rather than doing nothing
if (queue->flags & WORK_QUEUE_FLAG_MULTI)
{
- work_thread_info *thread = &queue->thread[queue->threads];
+ work_thread_info *thread = queue->thread[queue->threads];
end_timing(thread->waittime);
@@ -341,10 +362,10 @@ int osd_work_queue_wait(osd_work_queue *queue, osd_ticks_t timeout)
}
// reset our done event and double-check the items before waiting
- osd_event_reset(queue->doneevent);
+ queue->doneevent.reset();
queue->waiting = TRUE;
if (queue->items != 0)
- osd_event_wait(queue->doneevent, timeout);
+ queue->doneevent.wait(timeout);
queue->waiting = FALSE;
// return TRUE if we actually hit 0
@@ -358,73 +379,53 @@ int osd_work_queue_wait(osd_work_queue *queue, osd_ticks_t timeout)
void osd_work_queue_free(osd_work_queue *queue)
{
- // if we have threads, clean them up
- if (queue->thread != NULL)
+ // stop the timer for "waittime" on the main thread
+ if (queue->flags & WORK_QUEUE_FLAG_MULTI)
{
- int threadnum;
+ end_timing(queue->thread[queue->threads]->waittime);
+ }
- // stop the timer for "waittime" on the main thread
- if (queue->flags & WORK_QUEUE_FLAG_MULTI)
- {
- end_timing(queue->thread[queue->threads].waittime);
- }
+ // signal all the threads to exit
+ queue->exiting = TRUE;
+ for (int threadnum = 0; threadnum < queue->threads; threadnum++)
+ {
+ work_thread_info *thread = queue->thread[threadnum];
+ thread->wakeevent.set();
+ }
- // signal all the threads to exit
- queue->exiting = TRUE;
- for (threadnum = 0; threadnum < queue->threads; threadnum++)
- {
- work_thread_info *thread = &queue->thread[threadnum];
- if (thread->wakeevent != NULL)
- osd_event_set(thread->wakeevent);
- }
+ // wait for all the threads to go away
+ for (int threadnum = 0; threadnum < queue->threads; threadnum++)
+ {
+ work_thread_info *thread = queue->thread[threadnum];
- // wait for all the threads to go away
- for (threadnum = 0; threadnum < queue->threads; threadnum++)
+ // block on the thread going away, then close the handle
+ if (thread->handle != NULL)
{
- work_thread_info *thread = &queue->thread[threadnum];
-
- // block on the thread going away, then close the handle
- if (thread->handle != NULL)
- {
- thread->handle->join();
- delete thread->handle;
- }
-
- // clean up the wake event
- if (thread->wakeevent != NULL)
- osd_event_free(thread->wakeevent);
+ thread->handle->join();
+ delete thread->handle;
}
-#if KEEP_STATISTICS
- int allocthreadnum;
- if (queue->flags & WORK_QUEUE_FLAG_MULTI)
- allocthreadnum = queue->threads + 1;
- else
- allocthreadnum = queue->threads;
+ }
- // output per-thread statistics
- for (threadnum = 0; threadnum < allocthreadnum; threadnum++)
- {
- work_thread_info *thread = &queue->thread[threadnum];
- osd_ticks_t total = thread->runtime + thread->waittime + thread->spintime;
- printf("Thread %d: items=%9d run=%5.2f%% (%5.2f%%) spin=%5.2f%% wait/other=%5.2f%% total=%9d\n",
- threadnum, thread->itemsdone,
- (double)thread->runtime * 100.0 / (double)total,
- (double)thread->actruntime * 100.0 / (double)total,
- (double)thread->spintime * 100.0 / (double)total,
- (double)thread->waittime * 100.0 / (double)total,
- (UINT32) total);
- }
-#endif
+#if KEEP_STATISTICS
+ // output per-thread statistics
+ for (work_thread_info *thread : queue->thread)
+ {
+ osd_ticks_t total = thread->runtime + thread->waittime + thread->spintime;
+ printf("Thread %d: items=%9d run=%5.2f%% (%5.2f%%) spin=%5.2f%% wait/other=%5.2f%% total=%9d\n",
+ thread->id, thread->itemsdone,
+ (double)thread->runtime * 100.0 / (double)total,
+ (double)thread->actruntime * 100.0 / (double)total,
+ (double)thread->spintime * 100.0 / (double)total,
+ (double)thread->waittime * 100.0 / (double)total,
+ (UINT32) total);
}
+#endif
// free the list
- if (queue->thread != NULL)
- osd_free(queue->thread);
-
- // free all the events
- if (queue->doneevent != NULL)
- osd_event_free(queue->doneevent);
+ for (auto & th : queue->thread)
+ delete th;
+ queue->thread.clear();
// free all items in the free list
while (queue->free.load() != nullptr)
@@ -432,8 +433,8 @@ void osd_work_queue_free(osd_work_queue *queue)
osd_work_item *item = (osd_work_item *)queue->free;
queue->free = item->next;
if (item->event != NULL)
- osd_event_free(item->event);
- osd_free(item);
+ delete item->event;
+ delete item;
}
// free all items in the active list
@@ -442,20 +443,19 @@ void osd_work_queue_free(osd_work_queue *queue)
osd_work_item *item = (osd_work_item *)queue->list;
queue->list = item->next;
if (item->event != NULL)
- osd_event_free(item->event);
- osd_free(item);
+ delete item->event;
+ delete item;
}
#if KEEP_STATISTICS
- printf("Items queued = %9d\n", queue->itemsqueued);
- printf("SetEvent calls = %9d\n", queue->setevents);
- printf("Extra items = %9d\n", queue->extraitems);
- printf("Spin loops = %9d\n", queue->spinloops);
+ printf("Items queued = %9d\n", queue->itemsqueued.load());
+ printf("SetEvent calls = %9d\n", queue->setevents.load());
+ printf("Extra items = %9d\n", queue->extraitems.load());
+ printf("Spin loops = %9d\n", queue->spinloops.load());
#endif
- delete queue->lock;
// free the queue itself
- osd_free(queue);
+ delete queue;
}
@@ -476,7 +476,7 @@ osd_work_item *osd_work_item_queue_multiple(osd_work_queue *queue, osd_work_call
// first allocate a new work item; try the free list first
{
- std::lock_guard<std::mutex> lock(*queue->lock);
+ std::lock_guard<std::mutex> lock(queue->lock);
do
{
item = (osd_work_item *)queue->free;
@@ -487,12 +487,9 @@ osd_work_item *osd_work_item_queue_multiple(osd_work_queue *queue, osd_work_call
if (item == NULL)
{
// allocate the item
- item = (osd_work_item *)osd_malloc(sizeof(*item));
+ item = new osd_work_item(*queue);
if (item == NULL)
return NULL;
- item->event = NULL;
- item->queue = queue;
- item->done = FALSE;
}
else
{
@@ -515,7 +512,7 @@ osd_work_item *osd_work_item_queue_multiple(osd_work_queue *queue, osd_work_call
// enqueue the whole thing within the critical section
{
- std::lock_guard<std::mutex> lock(*queue->lock);
+ std::lock_guard<std::mutex> lock(queue->lock);
*queue->tailptr = itemlist;
queue->tailptr = item_tailptr;
}
@@ -532,12 +529,12 @@ osd_work_item *osd_work_item_queue_multiple(osd_work_queue *queue, osd_work_call
// iterate over all the threads
for (threadnum = 0; threadnum < queue->threads; threadnum++)
{
- work_thread_info *thread = &queue->thread[threadnum];
+ work_thread_info *thread = queue->thread[threadnum];
// if this thread is not active, wake him up
if (!thread->active)
{
- osd_event_set(thread->wakeevent);
+ thread->wakeevent.set();
add_to_stat(queue->setevents, 1);
// for non-shared, the first one we find is good enough
@@ -550,9 +547,9 @@ osd_work_item *osd_work_item_queue_multiple(osd_work_queue *queue, osd_work_call
// if no threads, run the queue now on this thread
if (queue->threads == 0)
{
- end_timing(queue->thread[0].waittime);
- worker_thread_process(queue, &queue->thread[0]);
- begin_timing(queue->thread[0].waittime);
+ end_timing(queue->thread[0]->waittime);
+ worker_thread_process(queue, queue->thread[0]);
+ begin_timing(queue->thread[0]->waittime);
}
// only return the item if it won't get released automatically
return (flags & WORK_ITEM_FLAG_AUTO_RELEASE) ? NULL : lastitem;
@@ -572,11 +569,11 @@ int osd_work_item_wait(osd_work_item *item, osd_ticks_t timeout)
// if we don't have an event, create one
if (item->event == NULL)
{
- std::lock_guard<std::mutex> lock(*item->queue->lock);
- item->event = osd_event_alloc(TRUE, FALSE); // manual reset, not signalled
+ std::lock_guard<std::mutex> lock(item->queue.lock);
+ item->event = new osd_event(TRUE, FALSE); // manual reset, not signalled
}
else
- osd_event_reset(item->event);
+ item->event->reset();
// if we don't have an event, we need to spin (shouldn't ever really happen)
if (item->event == NULL)
@@ -587,7 +584,7 @@ int osd_work_item_wait(osd_work_item *item, osd_ticks_t timeout)
// otherwise, block on the event until done
else if (!item->done)
- osd_event_wait(item->event, timeout);
+ item->event->wait(timeout);
// return TRUE if the refcount actually hit 0
return item->done;
@@ -616,12 +613,12 @@ void osd_work_item_release(osd_work_item *item)
osd_work_item_wait(item, 100 * osd_ticks_per_second());
// add us to the free list on our queue
- std::lock_guard<std::mutex> lock(*item->queue->lock);
+ std::lock_guard<std::mutex> lock(item->queue.lock);
do
{
- next = (osd_work_item *)item->queue->free;
+ next = (osd_work_item *) item->queue.free;
item->next = next;
- } while (!item->queue->free.compare_exchange_weak(next, item, std::memory_order_release, std::memory_order_relaxed));
+ } while (!item->queue.free.compare_exchange_weak(next, item, std::memory_order_release, std::memory_order_relaxed));
}
@@ -661,64 +658,56 @@ static int effective_num_processors(void)
static void *worker_thread_entry(void *param)
{
work_thread_info *thread = (work_thread_info *)param;
- osd_work_queue *queue = thread->queue;
-
-#if defined(SDLMAME_MACOSX)
- void *arp = NewAutoreleasePool();
-#endif
+ osd_work_queue &queue = thread->queue;
// loop until we exit
for ( ;; )
{
// block waiting for work or exit
// bail on exit, and only wait if there are no pending items in queue
- if (queue->exiting)
+ if (queue.exiting)
break;
- if (!queue_has_list_items(queue))
+ if (!queue_has_list_items(&queue))
{
begin_timing(thread->waittime);
- osd_event_wait(thread->wakeevent, OSD_EVENT_WAIT_INFINITE);
+ thread->wakeevent.wait( OSD_EVENT_WAIT_INFINITE);
end_timing(thread->waittime);
}
- if (queue->exiting)
+ if (queue.exiting)
break;
// indicate that we are live
thread->active = TRUE;
- ++queue->livethreads;
+ ++queue.livethreads;
// process work items
for ( ;; )
{
// process as much as we can
- worker_thread_process(queue, thread);
+ worker_thread_process(&queue, thread);
// if we're a high frequency queue, spin for a while before giving up
- if (queue->flags & WORK_QUEUE_FLAG_HIGH_FREQ && queue->list.load() == nullptr)
+ if (queue.flags & WORK_QUEUE_FLAG_HIGH_FREQ && queue.list.load() == nullptr)
{
// spin for a while looking for more work
begin_timing(thread->spintime);
- spin_while<std::atomic<osd_work_item *>, osd_work_item *>(&queue->list, (osd_work_item *)nullptr, SPIN_LOOP_TIME);
+ spin_while<std::atomic<osd_work_item *>, osd_work_item *>(&queue.list, (osd_work_item *)nullptr, SPIN_LOOP_TIME);
end_timing(thread->spintime);
}
// if nothing more, release the processor
- if (!queue_has_list_items(queue))
+ if (!queue_has_list_items(&queue))
break;
- add_to_stat(queue->spinloops, 1);
+ add_to_stat(queue.spinloops, 1);
}
// decrement the live thread count
thread->active = FALSE;
- --queue->livethreads;
+ --queue.livethreads;
}
-#if defined(SDLMAME_MACOSX)
- ReleaseAutoreleasePool(arp);
-#endif
-
return NULL;
}
@@ -729,7 +718,7 @@ static void *worker_thread_entry(void *param)
static void worker_thread_process(osd_work_queue *queue, work_thread_info *thread)
{
- int threadid = thread - queue->thread;
+ int threadid = thread->id;
begin_timing(thread->runtime);
@@ -742,7 +731,7 @@ static void worker_thread_process(osd_work_queue *queue, work_thread_info *threa
// use a critical section to synchronize the removal of items
{
- std::lock_guard<std::mutex> lock(*queue->lock);
+ std::lock_guard<std::mutex> lock(queue->lock);
if (queue->list.load() == nullptr)
{
@@ -784,12 +773,12 @@ static void worker_thread_process(osd_work_queue *queue, work_thread_info *threa
// set the result and signal the event
else
{
- std::lock_guard<std::mutex> lock(*queue->lock);
+ std::lock_guard<std::mutex> lock(queue->lock);
if (item->event != NULL)
{
- osd_event_set(item->event);
- add_to_stat(item->queue->setevents, 1);
+ item->event->set();
+ add_to_stat(item->queue.setevents, 1);
}
}
@@ -802,7 +791,7 @@ static void worker_thread_process(osd_work_queue *queue, work_thread_info *threa
// we don't need to set the doneevent for multi queues because they spin
if (queue->waiting)
{
- osd_event_set(queue->doneevent);
+ queue->doneevent.set();
add_to_stat(queue->setevents, 1);
}
@@ -811,7 +800,7 @@ static void worker_thread_process(osd_work_queue *queue, work_thread_info *threa
bool queue_has_list_items(osd_work_queue *queue)
{
- std::lock_guard<std::mutex> lock(*queue->lock);
+ std::lock_guard<std::mutex> lock(queue->lock);
bool has_list_items = (queue->list.load() != nullptr);
return has_list_items;
}