ref: e656f44c241293134445496c3c005a0e543eb022
parent: 8c6263e8dc9e1656057e51d7cd916b4db26b1aea
author: James Zern <[email protected]>
date: Thu Jun 19 17:14:51 EDT 2014
update vp9_thread.[hc] pull the latest from WebP, which adds a worker interface abstraction allowing an application to override init/reset/sync/launch/execute/end this has the side effect of removing a harmless, but annoying, TSan warning. Original source: http://git.chromium.org/webm/libwebp.git 100644 blob 08ad4e1fecba302bf1247645e84a7d2779956bc3 src/utils/thread.c 100644 blob 7bd451b124ae3b81596abfbcc823e3cb129d3a38 src/utils/thread.h Local modifications: - s/WebP/VP9/g - camelcase functions -> lower with _'s - associate '*' with the variable, not the type Change-Id: I875ac5a74ed873cbcb19a3a100b5e0ca6fcd9aed
--- a/test/vp9_thread_test.cc
+++ b/test/vp9_thread_test.cc
@@ -28,11 +28,11 @@
protected:
virtual ~VP9WorkerThreadTest() {} virtual void SetUp() {- vp9_worker_init(&worker_);
+ vp9_get_worker_interface()->init(&worker_);
}
virtual void TearDown() {- vp9_worker_end(&worker_);
+ vp9_get_worker_interface()->end(&worker_);
}
VP9Worker worker_;
@@ -45,10 +45,11 @@
}
TEST_P(VP9WorkerThreadTest, HookSuccess) {- EXPECT_NE(vp9_worker_sync(&worker_), 0); // should be a no-op.
+ // should be a no-op.
+ EXPECT_NE(vp9_get_worker_interface()->sync(&worker_), 0);
for (int i = 0; i < 2; ++i) {- EXPECT_NE(vp9_worker_reset(&worker_), 0);
+ EXPECT_NE(vp9_get_worker_interface()->reset(&worker_), 0);
int hook_data = 0;
int return_value = 1; // return successfully from the hook
@@ -58,20 +59,21 @@
const bool synchronous = GetParam();
if (synchronous) {- vp9_worker_execute(&worker_);
+ vp9_get_worker_interface()->execute(&worker_);
} else {- vp9_worker_launch(&worker_);
+ vp9_get_worker_interface()->launch(&worker_);
}
- EXPECT_NE(vp9_worker_sync(&worker_), 0);
+ EXPECT_NE(vp9_get_worker_interface()->sync(&worker_), 0);
EXPECT_FALSE(worker_.had_error);
EXPECT_EQ(5, hook_data);
- EXPECT_NE(vp9_worker_sync(&worker_), 0); // should be a no-op.
+ // should be a no-op.
+ EXPECT_NE(vp9_get_worker_interface()->sync(&worker_), 0);
}
}
TEST_P(VP9WorkerThreadTest, HookFailure) {- EXPECT_NE(vp9_worker_reset(&worker_), 0);
+ EXPECT_NE(vp9_get_worker_interface()->reset(&worker_), 0);
int hook_data = 0;
int return_value = 0; // return failure from the hook
@@ -81,22 +83,40 @@
const bool synchronous = GetParam();
if (synchronous) {- vp9_worker_execute(&worker_);
+ vp9_get_worker_interface()->execute(&worker_);
} else {- vp9_worker_launch(&worker_);
+ vp9_get_worker_interface()->launch(&worker_);
}
- EXPECT_FALSE(vp9_worker_sync(&worker_));
+ EXPECT_FALSE(vp9_get_worker_interface()->sync(&worker_));
EXPECT_EQ(1, worker_.had_error);
// Ensure _reset() clears the error and _launch() can be called again.
return_value = 1;
- EXPECT_NE(vp9_worker_reset(&worker_), 0);
+ EXPECT_NE(vp9_get_worker_interface()->reset(&worker_), 0);
EXPECT_FALSE(worker_.had_error);
- vp9_worker_launch(&worker_);
- EXPECT_NE(vp9_worker_sync(&worker_), 0);
+ vp9_get_worker_interface()->launch(&worker_);
+ EXPECT_NE(vp9_get_worker_interface()->sync(&worker_), 0);
EXPECT_FALSE(worker_.had_error);
}
+TEST(VP9WorkerThreadTest, TestInterfaceAPI) {+ EXPECT_EQ(0, vp9_set_worker_interface(NULL));
+ EXPECT_TRUE(vp9_get_worker_interface() != NULL);
+ for (int i = 0; i < 6; ++i) {+ VP9WorkerInterface winterface = *vp9_get_worker_interface();
+ switch (i) {+ default:
+ case 0: winterface.init = NULL; break;
+ case 1: winterface.reset = NULL; break;
+ case 2: winterface.sync = NULL; break;
+ case 3: winterface.launch = NULL; break;
+ case 4: winterface.execute = NULL; break;
+ case 5: winterface.end = NULL; break;
+ }
+ EXPECT_EQ(0, vp9_set_worker_interface(&winterface));
+ }
+}
+
// -----------------------------------------------------------------------------
// Multi-threaded decode tests
@@ -143,6 +163,44 @@
<< "threads = " << t;
}
}
+}
+
+// Trivial serialized thread worker interface implementation.
+// Note any worker that requires synchronization between other workers will
+// hang.
+namespace impl {+
+void Init(VP9Worker *const worker) { memset(worker, 0, sizeof(*worker)); }+int Reset(VP9Worker *const /*worker*/) { return 1; }+int Sync(VP9Worker *const worker) { return !worker->had_error; }+
+void Execute(VP9Worker *const worker) {+ worker->had_error |= worker->hook(worker->data1, worker->data2);
+}
+
+void Launch(VP9Worker *const worker) { Execute(worker); }+void End(VP9Worker *const /*worker*/) {}+
+} // namespace impl
+
+TEST(VP9WorkerThreadTest, TestSerialInterface) {+ static const VP9WorkerInterface serial_interface = {+ impl::Init, impl::Reset, impl::Sync, impl::Launch, impl::Execute, impl::End
+ };
+ // TODO(jzern): Avoid using a file that will use the row-based thread
+ // loopfilter, with the simple serialized implementation it will hang. This is
+ // due to its expectation that rows will be run in parallel as they wait on
+ // progress in the row above before proceeding.
+ static const char expected_md5[] = "b35a1b707b28e82be025d960aba039bc";
+ static const char filename[] = "vp90-2-03-size-226x226.webm";
+ VP9WorkerInterface default_interface = *vp9_get_worker_interface();
+
+ EXPECT_NE(vp9_set_worker_interface(&serial_interface), 0);
+ EXPECT_EQ(expected_md5, DecodeFile(filename, 2));
+
+ // Reset the interface.
+ EXPECT_NE(vp9_set_worker_interface(&default_interface), 0);
+ EXPECT_EQ(expected_md5, DecodeFile(filename, 2));
}
TEST(VP9DecodeMultiThreadedTest, Decode) {--- a/vp9/decoder/vp9_decodeframe.c
+++ b/vp9/decoder/vp9_decodeframe.c
@@ -767,6 +767,7 @@
const uint8_t *data,
const uint8_t *data_end) {VP9_COMMON *const cm = &pbi->common;
+ const VP9WorkerInterface *const winterface = vp9_get_worker_interface();
const int aligned_cols = mi_cols_aligned_to_sb(cm->mi_cols);
const int tile_cols = 1 << cm->log2_tile_cols;
const int tile_rows = 1 << cm->log2_tile_rows;
@@ -779,7 +780,7 @@
CHECK_MEM_ERROR(cm, pbi->lf_worker.data1,
vpx_memalign(32, sizeof(LFWorkerData)));
pbi->lf_worker.hook = (VP9WorkerHook)vp9_loop_filter_worker;
- if (pbi->max_threads > 1 && !vp9_worker_reset(&pbi->lf_worker)) {+ if (pbi->max_threads > 1 && !winterface->reset(&pbi->lf_worker)) {vpx_internal_error(&cm->error, VPX_CODEC_ERROR,
"Loop filter thread creation failed");
}
@@ -865,13 +866,13 @@
// decoding has completed: finish up the loop filter in this thread.
if (mi_row + MI_BLOCK_SIZE >= cm->mi_rows) continue;
- vp9_worker_sync(&pbi->lf_worker);
+ winterface->sync(&pbi->lf_worker);
lf_data->start = lf_start;
lf_data->stop = mi_row;
if (pbi->max_threads > 1) {- vp9_worker_launch(&pbi->lf_worker);
+ winterface->launch(&pbi->lf_worker);
} else {- vp9_worker_execute(&pbi->lf_worker);
+ winterface->execute(&pbi->lf_worker);
}
}
}
@@ -880,10 +881,10 @@
// Loopfilter remaining rows in the frame.
if (cm->lf.filter_level) {LFWorkerData *const lf_data = (LFWorkerData*)pbi->lf_worker.data1;
- vp9_worker_sync(&pbi->lf_worker);
+ winterface->sync(&pbi->lf_worker);
lf_data->start = lf_data->stop;
lf_data->stop = cm->mi_rows;
- vp9_worker_execute(&pbi->lf_worker);
+ winterface->execute(&pbi->lf_worker);
}
// Get last tile data.
@@ -927,6 +928,7 @@
const uint8_t *data,
const uint8_t *data_end) {VP9_COMMON *const cm = &pbi->common;
+ const VP9WorkerInterface *const winterface = vp9_get_worker_interface();
const uint8_t *bit_reader_end = NULL;
const int aligned_mi_cols = mi_cols_aligned_to_sb(cm->mi_cols);
const int tile_cols = 1 << cm->log2_tile_cols;
@@ -953,11 +955,11 @@
VP9Worker *const worker = &pbi->tile_workers[i];
++pbi->num_tile_workers;
- vp9_worker_init(worker);
+ winterface->init(worker);
CHECK_MEM_ERROR(cm, worker->data1,
vpx_memalign(32, sizeof(TileWorkerData)));
CHECK_MEM_ERROR(cm, worker->data2, vpx_malloc(sizeof(TileInfo)));
- if (i < num_threads - 1 && !vp9_worker_reset(worker)) {+ if (i < num_threads - 1 && !winterface->reset(worker)) {vpx_internal_error(&cm->error, VPX_CODEC_ERROR,
"Tile decoder thread creation failed");
}
@@ -1020,9 +1022,9 @@
worker->had_error = 0;
if (i == num_workers - 1 || n == tile_cols - 1) {- vp9_worker_execute(worker);
+ winterface->execute(worker);
} else {- vp9_worker_launch(worker);
+ winterface->launch(worker);
}
if (buf->col == tile_cols - 1) {@@ -1034,7 +1036,7 @@
for (; i > 0; --i) {VP9Worker *const worker = &pbi->tile_workers[i - 1];
- pbi->mb.corrupted |= !vp9_worker_sync(worker);
+ pbi->mb.corrupted |= !winterface->sync(worker);
}
if (final_worker > -1) {TileWorkerData *const tile_data =
--- a/vp9/decoder/vp9_decoder.c
+++ b/vp9/decoder/vp9_decoder.c
@@ -76,7 +76,7 @@
cm->error.setjmp = 0;
- vp9_worker_init(&pbi->lf_worker);
+ vp9_get_worker_interface()->init(&pbi->lf_worker);
return pbi;
}
@@ -86,12 +86,12 @@
int i;
vp9_remove_common(cm);
- vp9_worker_end(&pbi->lf_worker);
+ vp9_get_worker_interface()->end(&pbi->lf_worker);
vpx_free(pbi->lf_worker.data1);
vpx_free(pbi->tile_data);
for (i = 0; i < pbi->num_tile_workers; ++i) {VP9Worker *const worker = &pbi->tile_workers[i];
- vp9_worker_end(worker);
+ vp9_get_worker_interface()->end(worker);
vpx_free(worker->data1);
vpx_free(worker->data2);
}
--- a/vp9/decoder/vp9_dthread.c
+++ b/vp9/decoder/vp9_dthread.c
@@ -138,6 +138,7 @@
int frame_filter_level,
int y_only) {VP9LfSync *const lf_sync = &pbi->lf_row_sync;
+ const VP9WorkerInterface *const winterface = vp9_get_worker_interface();
// Number of superblock rows and cols
const int sb_rows = mi_cols_aligned_to_sb(cm->mi_rows) >> MI_BLOCK_SIZE_LOG2;
const int tile_cols = 1 << cm->log2_tile_cols;
@@ -197,15 +198,15 @@
// Start loopfiltering
if (i == num_workers - 1) {- vp9_worker_execute(worker);
+ winterface->execute(worker);
} else {- vp9_worker_launch(worker);
+ winterface->launch(worker);
}
}
// Wait till all rows are finished
for (i = 0; i < num_workers; ++i) {- vp9_worker_sync(&pbi->tile_workers[i]);
+ winterface->sync(&pbi->tile_workers[i]);
}
}
--- a/vp9/decoder/vp9_thread.c
+++ b/vp9/decoder/vp9_thread.c
@@ -11,59 +11,67 @@
//
// Original source:
// http://git.chromium.org/webm/libwebp.git
-// 100644 blob eff8f2a8c20095aade3c292b0e9292dac6cb3587 src/utils/thread.c
+// 100644 blob 08ad4e1fecba302bf1247645e84a7d2779956bc3 src/utils/thread.c
-
#include <assert.h>
#include <string.h> // for memset()
#include "./vp9_thread.h"
+#include "vpx_mem/vpx_mem.h"
-#if defined(__cplusplus) || defined(c_plusplus)
-extern "C" {-#endif
-
#if CONFIG_MULTITHREAD
+struct VP9WorkerImpl {+ pthread_mutex_t mutex_;
+ pthread_cond_t condition_;
+ pthread_t thread_;
+};
+
//------------------------------------------------------------------------------
-static THREADFN thread_loop(void *ptr) { // thread loop- VP9Worker* const worker = (VP9Worker*)ptr;
+static void execute(VP9Worker *const worker); // Forward declaration.
+
+static THREADFN thread_loop(void *ptr) {+ VP9Worker *const worker = (VP9Worker*)ptr;
int done = 0;
while (!done) {- pthread_mutex_lock(&worker->mutex_);
+ pthread_mutex_lock(&worker->impl_->mutex_);
while (worker->status_ == OK) { // wait in idling mode- pthread_cond_wait(&worker->condition_, &worker->mutex_);
+ pthread_cond_wait(&worker->impl_->condition_, &worker->impl_->mutex_);
}
if (worker->status_ == WORK) {- vp9_worker_execute(worker);
+ execute(worker);
worker->status_ = OK;
} else if (worker->status_ == NOT_OK) { // finish the workerdone = 1;
}
- // signal to the main thread that we're done (for Sync())
- pthread_cond_signal(&worker->condition_);
- pthread_mutex_unlock(&worker->mutex_);
+ // signal to the main thread that we're done (for sync())
+ pthread_cond_signal(&worker->impl_->condition_);
+ pthread_mutex_unlock(&worker->impl_->mutex_);
}
return THREAD_RETURN(NULL); // Thread is finished
}
// main thread state control
-static void change_state(VP9Worker* const worker,
+static void change_state(VP9Worker *const worker,
VP9WorkerStatus new_status) {- // no-op when attempting to change state on a thread that didn't come up
- if (worker->status_ < OK) return;
+ // No-op when attempting to change state on a thread that didn't come up.
+ // Checking status_ without acquiring the lock first would result in a data
+ // race.
+ if (worker->impl_ == NULL) return;
- pthread_mutex_lock(&worker->mutex_);
- // wait for the worker to finish
- while (worker->status_ != OK) {- pthread_cond_wait(&worker->condition_, &worker->mutex_);
+ pthread_mutex_lock(&worker->impl_->mutex_);
+ if (worker->status_ >= OK) {+ // wait for the worker to finish
+ while (worker->status_ != OK) {+ pthread_cond_wait(&worker->impl_->condition_, &worker->impl_->mutex_);
+ }
+ // assign new status and release the working thread if needed
+ if (new_status != OK) {+ worker->status_ = new_status;
+ pthread_cond_signal(&worker->impl_->condition_);
+ }
}
- // assign new status and release the working thread if needed
- if (new_status != OK) {- worker->status_ = new_status;
- pthread_cond_signal(&worker->condition_);
- }
- pthread_mutex_unlock(&worker->mutex_);
+ pthread_mutex_unlock(&worker->impl_->mutex_);
}
#endif // CONFIG_MULTITHREAD
@@ -70,12 +78,12 @@
//------------------------------------------------------------------------------
-void vp9_worker_init(VP9Worker* const worker) {+static void init(VP9Worker *const worker) {memset(worker, 0, sizeof(*worker));
worker->status_ = NOT_OK;
}
-int vp9_worker_sync(VP9Worker* const worker) {+static int sync(VP9Worker *const worker) {#if CONFIG_MULTITHREAD
change_state(worker, OK);
#endif
@@ -83,59 +91,93 @@
return !worker->had_error;
}
-int vp9_worker_reset(VP9Worker* const worker) {+static int reset(VP9Worker *const worker) {int ok = 1;
worker->had_error = 0;
if (worker->status_ < OK) {#if CONFIG_MULTITHREAD
- if (pthread_mutex_init(&worker->mutex_, NULL) ||
- pthread_cond_init(&worker->condition_, NULL)) {+ worker->impl_ = (VP9WorkerImpl*)vpx_calloc(1, sizeof(*worker->impl_));
+ if (worker->impl_ == NULL) {return 0;
}
- pthread_mutex_lock(&worker->mutex_);
- ok = !pthread_create(&worker->thread_, NULL, thread_loop, worker);
+ if (pthread_mutex_init(&worker->impl_->mutex_, NULL)) {+ goto Error;
+ }
+ if (pthread_cond_init(&worker->impl_->condition_, NULL)) {+ pthread_mutex_destroy(&worker->impl_->mutex_);
+ goto Error;
+ }
+ pthread_mutex_lock(&worker->impl_->mutex_);
+ ok = !pthread_create(&worker->impl_->thread_, NULL, thread_loop, worker);
if (ok) worker->status_ = OK;
- pthread_mutex_unlock(&worker->mutex_);
+ pthread_mutex_unlock(&worker->impl_->mutex_);
+ if (!ok) {+ pthread_mutex_destroy(&worker->impl_->mutex_);
+ pthread_cond_destroy(&worker->impl_->condition_);
+ Error:
+ vpx_free(worker->impl_);
+ worker->impl_ = NULL;
+ return 0;
+ }
#else
worker->status_ = OK;
#endif
} else if (worker->status_ > OK) {- ok = vp9_worker_sync(worker);
+ ok = sync(worker);
}
assert(!ok || (worker->status_ == OK));
return ok;
}
-void vp9_worker_execute(VP9Worker* const worker) {+static void execute(VP9Worker *const worker) { if (worker->hook != NULL) {worker->had_error |= !worker->hook(worker->data1, worker->data2);
}
}
-void vp9_worker_launch(VP9Worker* const worker) {+static void launch(VP9Worker *const worker) {#if CONFIG_MULTITHREAD
change_state(worker, WORK);
#else
- vp9_worker_execute(worker);
+ execute(worker);
#endif
}
-void vp9_worker_end(VP9Worker* const worker) {+static void end(VP9Worker *const worker) { if (worker->status_ >= OK) {#if CONFIG_MULTITHREAD
change_state(worker, NOT_OK);
- pthread_join(worker->thread_, NULL);
- pthread_mutex_destroy(&worker->mutex_);
- pthread_cond_destroy(&worker->condition_);
+ pthread_join(worker->impl_->thread_, NULL);
+ pthread_mutex_destroy(&worker->impl_->mutex_);
+ pthread_cond_destroy(&worker->impl_->condition_);
#else
worker->status_ = NOT_OK;
#endif
}
+ vpx_free(worker->impl_);
+ worker->impl_ = NULL;
assert(worker->status_ == NOT_OK);
}
//------------------------------------------------------------------------------
-#if defined(__cplusplus) || defined(c_plusplus)
-} // extern "C"
-#endif
+static VP9WorkerInterface g_worker_interface = {+ init, reset, sync, launch, execute, end
+};
+
+int vp9_set_worker_interface(const VP9WorkerInterface* const winterface) {+ if (winterface == NULL ||
+ winterface->init == NULL || winterface->reset == NULL ||
+ winterface->sync == NULL || winterface->launch == NULL ||
+ winterface->execute == NULL || winterface->end == NULL) {+ return 0;
+ }
+ g_worker_interface = *winterface;
+ return 1;
+}
+
+const VP9WorkerInterface *vp9_get_worker_interface(void) {+ return &g_worker_interface;
+}
+
+//------------------------------------------------------------------------------
--- a/vp9/decoder/vp9_thread.h
+++ b/vp9/decoder/vp9_thread.h
@@ -11,9 +11,8 @@
//
// Original source:
// http://git.chromium.org/webm/libwebp.git
-// 100644 blob 13a61a4c84194c3374080cbf03d881d3cd6af40d src/utils/thread.h
+// 100644 blob 7bd451b124ae3b81596abfbcc823e3cb129d3a38 src/utils/thread.h
-
#ifndef VP9_DECODER_VP9_THREAD_H_
#define VP9_DECODER_VP9_THREAD_H_
@@ -163,40 +162,53 @@
// arguments (data1 and data2), and should return false in case of error.
typedef int (*VP9WorkerHook)(void*, void*);
-// Synchronize object used to launch job in the worker thread
+// Platform-dependent implementation details for the worker.
+typedef struct VP9WorkerImpl VP9WorkerImpl;
+
+// Synchronization object used to launch job in the worker thread
typedef struct {-#if CONFIG_MULTITHREAD
- pthread_mutex_t mutex_;
- pthread_cond_t condition_;
- pthread_t thread_;
-#endif
+ VP9WorkerImpl *impl_;
VP9WorkerStatus status_;
VP9WorkerHook hook; // hook to call
- void* data1; // first argument passed to 'hook'
- void* data2; // second argument passed to 'hook'
+ void *data1; // first argument passed to 'hook'
+ void *data2; // second argument passed to 'hook'
int had_error; // return value of the last call to 'hook'
} VP9Worker;
-// Must be called first, before any other method.
-void vp9_worker_init(VP9Worker* const worker);
-// Must be called to initialize the object and spawn the thread. Re-entrant.
-// Will potentially launch the thread. Returns false in case of error.
-int vp9_worker_reset(VP9Worker* const worker);
-// Makes sure the previous work is finished. Returns true if worker->had_error
-// was not set and no error condition was triggered by the working thread.
-int vp9_worker_sync(VP9Worker* const worker);
-// Triggers the thread to call hook() with data1 and data2 argument. These
-// hook/data1/data2 can be changed at any time before calling this function,
-// but not be changed afterward until the next call to vp9_worker_sync().
-void vp9_worker_launch(VP9Worker* const worker);
-// This function is similar to vp9_worker_launch() except that it calls the
-// hook directly instead of using a thread. Convenient to bypass the thread
-// mechanism while still using the VP9Worker structs. vp9_worker_sync() must
-// still be called afterward (for error reporting).
-void vp9_worker_execute(VP9Worker* const worker);
-// Kill the thread and terminate the object. To use the object again, one
-// must call vp9_worker_reset() again.
-void vp9_worker_end(VP9Worker* const worker);
+// The interface for all thread-worker related functions. All these functions
+// must be implemented.
+typedef struct {+ // Must be called first, before any other method.
+ void (*init)(VP9Worker *const worker);
+ // Must be called to initialize the object and spawn the thread. Re-entrant.
+ // Will potentially launch the thread. Returns false in case of error.
+ int (*reset)(VP9Worker *const worker);
+ // Makes sure the previous work is finished. Returns true if worker->had_error
+ // was not set and no error condition was triggered by the working thread.
+ int (*sync)(VP9Worker *const worker);
+ // Triggers the thread to call hook() with data1 and data2 arguments. These
+ // hook/data1/data2 values can be changed at any time before calling this
+ // function, but not be changed afterward until the next call to Sync().
+ void (*launch)(VP9Worker *const worker);
+ // This function is similar to launch() except that it calls the
+ // hook directly instead of using a thread. Convenient to bypass the thread
+ // mechanism while still using the VP9Worker structs. sync() must
+ // still be called afterward (for error reporting).
+ void (*execute)(VP9Worker *const worker);
+ // Kill the thread and terminate the object. To use the object again, one
+ // must call reset() again.
+ void (*end)(VP9Worker *const worker);
+} VP9WorkerInterface;
+
+// Install a new set of threading functions, overriding the defaults. This
+// should be done before any workers are started, i.e., before any encoding or
+// decoding takes place. The contents of the interface struct are copied, it
+// is safe to free the corresponding memory after this call. This function is
+// not thread-safe. Return false in case of invalid pointer or methods.
+int vp9_set_worker_interface(const VP9WorkerInterface *const winterface);
+
+// Retrieve the currently set thread worker interface.
+const VP9WorkerInterface *vp9_get_worker_interface(void);
//------------------------------------------------------------------------------