cleaned up lots of stuff (no functional changes)
ipc_stress_test runs all of the tests successfully now. In order to make
that work, I had to fix raw_queue_test so it didn't count the time spent
waiting for the fork() to complete as part of the timeout, and I cleaned
up various other parts at the same time.
Also improved some documentation and removed the mutex fairness tester
because it was slow, prone to random failures, and it's always been kind
of a fishy test.
diff --git a/aos/atom_code/ipc_lib/aos_sync.c b/aos/atom_code/ipc_lib/aos_sync.c
index 3a70411..1980a4d 100644
--- a/aos/atom_code/ipc_lib/aos_sync.c
+++ b/aos/atom_code/ipc_lib/aos_sync.c
@@ -173,7 +173,7 @@
}
}
-void condition_signal(mutex *c, mutex *m __attribute__((unused))) {
+void condition_signal(mutex *c) {
__sync_fetch_and_add(c, 1);
if (sys_futex(c, FUTEX_WAKE, 1, NULL, NULL, 0) == -1) {
fprintf(stderr, "sync: FUTEX_WAKE(%p, 1, NULL, NULL, 0)"
diff --git a/aos/atom_code/ipc_lib/aos_sync.h b/aos/atom_code/ipc_lib/aos_sync.h
index 9b28437..6d4bf55 100644
--- a/aos/atom_code/ipc_lib/aos_sync.h
+++ b/aos/atom_code/ipc_lib/aos_sync.h
@@ -47,32 +47,44 @@
// The futex_ functions are similar to the mutex_ ones but different.
// They are designed for signalling when something happens (possibly to
// multiple listeners). A mutex manipulated with them can only be set or unset.
+//
+// They are different from the condition_ functions in that they do NOT work
+// correctly as standard condition variables. While it is possible to keep
+// track of the "condition" using the value part of the futex_* functions, the
+// obvious implementation has basically the same race condition that condition
+// variables are designed to prevent between somebody else grabbing the mutex
+// and changing whether it's set or not and the futex_ function changing the
+// futex's value.
// Wait for the futex to be set. Will return immediately if it's already set.
-// Returns 0 if successful or it was already set, 1 if interrupted by a signal, or -1.
+// Returns 0 if successful or it was already set, 1 if interrupted by a signal,
+// or -1.
int futex_wait(mutex *m) __attribute__((warn_unused_result));
// Set the futex and wake up anybody waiting on it.
// Returns the number that were woken or -1.
//
-// This will always wake up all waiters at the same time.
+// This will always wake up all waiters at the same time and set the value to 1.
int futex_set(mutex *m);
// Same as above except lets something other than 1 be used as the final value.
int futex_set_value(mutex *m, mutex value);
-// Unsets the futex.
+// Unsets the futex (sets the value to 0).
// Returns 0 if it was set before and 1 if it wasn't.
// Can not fail.
int futex_unset(mutex *m);
// The condition_ functions implement condition variable support. The API is
// similar to the pthreads api and works the same way. The same m argument must
-// be passed in for all calls to all of the functions with a given c.
+// be passed in for all calls to all of the condition_ functions with a given c.
// Wait for the condition variable to be signalled. m will be unlocked
-// atomically with actually starting to wait.
+// atomically with actually starting to wait. m is guaranteed to be locked when
+// this function returns.
+// NOTE: The relocking of m is not atomic with stopping the actual wait and
+// other process(es) may lock (+unlock) the mutex first.
void condition_wait(mutex *c, mutex *m);
// If any other processes are condition_waiting on c, wake 1 of them. Does not
// require m to be locked.
-void condition_signal(mutex *c, mutex *m);
+void condition_signal(mutex *c);
// Wakes all processes that are condition_waiting on c. Does not require m to be
// locked.
void condition_broadcast(mutex *c, mutex *m);
diff --git a/aos/atom_code/ipc_lib/condition.cc b/aos/atom_code/ipc_lib/condition.cc
index c67ea5f..b764026 100644
--- a/aos/atom_code/ipc_lib/condition.cc
+++ b/aos/atom_code/ipc_lib/condition.cc
@@ -16,7 +16,7 @@
}
void Condition::Signal() {
- condition_signal(&impl_, &m_->impl_);
+ condition_signal(&impl_);
}
void Condition::Broadcast() {
diff --git a/aos/atom_code/ipc_lib/ipc_lib.gyp b/aos/atom_code/ipc_lib/ipc_lib.gyp
index c57dae7..f947d5e 100644
--- a/aos/atom_code/ipc_lib/ipc_lib.gyp
+++ b/aos/atom_code/ipc_lib/ipc_lib.gyp
@@ -49,7 +49,7 @@
],
},
{
- 'target_name': 'ipc_queue_test',
+ 'target_name': 'raw_queue_test',
'type': 'executable',
'sources': [
'queue_test.cc',
@@ -59,6 +59,8 @@
'queue',
'<(AOS)/build/aos.gyp:logging',
'core_lib',
+ '<(AOS)/common/common.gyp:queue_testutils',
+ '<(AOS)/common/common.gyp:time',
],
},
{
@@ -74,12 +76,6 @@
'<(AOS)/common/common.gyp:mutex',
'core_lib',
'<(AOS)/common/common.gyp:die',
-
- # These are the binaries that it runs.
- 'ipc_queue_test',
- '<(AOS)/common/common.gyp:queue_test',
- '<(AOS)/common/common.gyp:condition_test',
- '<(AOS)/common/common.gyp:mutex_test',
],
},
],
diff --git a/aos/atom_code/ipc_lib/ipc_stress_test.cc b/aos/atom_code/ipc_lib/ipc_stress_test.cc
index dc8c738..38c425f 100644
--- a/aos/atom_code/ipc_lib/ipc_stress_test.cc
+++ b/aos/atom_code/ipc_lib/ipc_stress_test.cc
@@ -5,6 +5,7 @@
#include <sys/types.h>
#include <sys/wait.h>
#include <libgen.h>
+#include <assert.h>
#include <vector>
#include <string>
@@ -21,34 +22,33 @@
// stderr output from each test run and only prints it out (not interleaved with
// the output from any other run) if the test fails.
//
-// It's written in C++ for performance. You need actual OS-level parallelism for
+// It's written in C++ for performance. We need actual OS-level parallelism for
// this to work, which means that Ruby's out because it doesn't have good
// support for doing that. My Python implementation ended up pretty heavily disk
-// IO-bound which isn't a very good way to test CPU contention.
+// IO-bound, which is a bad way to test CPU contention.
namespace aos {
// Each test is represented by the name of the test binary and then any
// arguments to pass to it.
// Using --gtest_filter is a bad idea because it seems to result in a lot of
-// swapping which causes everything to be disk-bound (at least on my computer).
+// swapping which causes everything to be disk-bound (at least for me).
static const ::std::vector< ::std::vector< ::std::string>> kTests = {
{"queue_test"},
{"condition_test"},
{"mutex_test"},
- // TODO(brians): Enable this one once it supports running in parallel.
- //{"ipc_queue_test"},
+ {"raw_queue_test"},
};
// These arguments get inserted before any per-test arguments.
static const ::std::vector< ::std::string> kDefaultArgs = {
- "--gtest_repeat=35",
+ "--gtest_repeat=30",
"--gtest_shuffle",
};
// How many test processes to run at a time.
-static const int kTesters = 12;
+static const int kTesters = 100;
// How long to test for.
-static constexpr time::Time kTestTime = time::Time::InSeconds(20);
+static constexpr time::Time kTestTime = time::Time::InSeconds(30);
// The structure that gets put into shared memory and then referenced by all of
// the child processes.
@@ -104,7 +104,7 @@
executable = ::std::string(shared->path) + "/" + c;
args[0] = executable.c_str();
for (const ::std::string &ci : kDefaultArgs) {
- args[i++] = ci.c_str();
+ args[++i] = ci.c_str();
}
} else {
args[i] = c.c_str();
diff --git a/aos/atom_code/ipc_lib/queue_test.cc b/aos/atom_code/ipc_lib/queue_test.cc
index 8925c65..9e5f3ae 100644
--- a/aos/atom_code/ipc_lib/queue_test.cc
+++ b/aos/atom_code/ipc_lib/queue_test.cc
@@ -10,29 +10,25 @@
#include "gtest/gtest.h"
-#include "aos/atom_code/ipc_lib/sharedmem_test_setup.h"
#include "aos/atom_code/ipc_lib/core_lib.h"
#include "aos/common/type_traits.h"
+#include "aos/common/queue_testutils.h"
+#include "aos/common/time.h"
+#include "aos/common/logging/logging.h"
using ::testing::AssertionResult;
using ::testing::AssertionSuccess;
using ::testing::AssertionFailure;
+using ::aos::common::testing::GlobalCoreInstance;
namespace aos {
namespace testing {
-// IMPORTANT: Some of the functions that do test predicate functions allocate
-// shared memory (and don't free it).
-class QueueTest : public SharedMemTestSetup {
+class QueueTest : public ::testing::Test {
protected:
static const size_t kFailureSize = 400;
static char *fatal_failure;
private:
- // This gets registered right after the fork, so it will get run before any
- // exit handlers that had already been registered.
- static void ExitExitHandler() {
- _exit(EXIT_SUCCESS);
- }
enum class ResultType : uint8_t {
NotCalled,
Called,
@@ -52,33 +48,35 @@
}
static_assert(aos::shm_ok<ResultType>::value,
"this will get put in shared memory");
- // Gets allocated in shared memory so it has to be volatile.
template<typename T>
struct FunctionToCall {
- ResultType result;
+ FunctionToCall() : result(ResultType::NotCalled) {
+ started.Lock();
+ }
+
+ volatile ResultType result;
bool expected;
void (*function)(T*, char*);
T *arg;
volatile char failure[kFailureSize];
+ Mutex started;
};
template<typename T>
- static void Hangs_(volatile FunctionToCall<T> *const to_call) {
+ static void Hangs_(FunctionToCall<T> *const to_call) {
+ to_call->started.Unlock();
to_call->result = ResultType::Called;
to_call->function(to_call->arg, const_cast<char *>(to_call->failure));
to_call->result = ResultType::Returned;
}
- static const long kMsToNs = 1000000;
- // The number of ms after which a function is considered to have hung.
- // Must be < 1000.
- static const long kHangTime = 10;
- static const unsigned int kForkSleep = 0; // how many seconds to sleep after forking
+ // How long until a function is considered to have hung.
+ static constexpr time::Time kHangTime = time::Time::InSeconds(0.035);
+ // How long to sleep after forking (for debugging).
+ static constexpr time::Time kForkSleep = time::Time::InSeconds(0);
// Represents a process that has been forked off. The destructor kills the
// process and wait(2)s for it.
class ForkedProcess {
- const pid_t pid_;
- mutex *const lock_;
public:
ForkedProcess(pid_t pid, mutex *lock) : pid_(pid), lock_(lock) {};
~ForkedProcess() {
@@ -93,24 +91,25 @@
}
const pid_t ret = wait(NULL);
if (ret == -1) {
- fprintf(stderr, "wait(NULL) failed."
- " child %jd might still be alive\n",
- static_cast<intmax_t>(pid_));
+ LOG(WARNING, "wait(NULL) failed."
+ " child %jd might still be alive\n",
+ static_cast<intmax_t>(pid_));
} else if (ret == 0) {
- fprintf(stderr, "child %jd wasn't waitable. it might still be alive\n",
- static_cast<intmax_t>(pid_));
+ LOG(WARNING, "child %jd wasn't waitable. it might still be alive\n",
+ static_cast<intmax_t>(pid_));
} else if (ret != pid_) {
- fprintf(stderr, "child %d is dead, but child %jd might still be alive\n",
- ret, static_cast<intmax_t>(pid_));
+ LOG(WARNING, "child %d is now confirmed dead"
+ ", but child %jd might still be alive\n",
+ ret, static_cast<intmax_t>(pid_));
}
}
enum class JoinResult {
Finished, Hung, Error
};
- JoinResult Join(long timeout = kHangTime) {
- timespec ts{kForkSleep, timeout * kMsToNs};
- switch (mutex_lock_timeout(lock_, &ts)) {
+ JoinResult Join(time::Time timeout = kHangTime) {
+ timespec lock_timeout = (kForkSleep + timeout).ToTimespec();
+ switch (mutex_lock_timeout(lock_, &lock_timeout)) {
case 2:
return JoinResult::Hung;
case 0:
@@ -119,9 +118,13 @@
return JoinResult::Error;
}
}
+
+ private:
+ const pid_t pid_;
+ mutex *const lock_;
} __attribute__((unused));
- // Member variables for HangsFork and HangsCheck.
+ // State for HangsFork and HangsCheck.
typedef uint8_t ChildID;
static void ReapExitHandler() {
for (auto it = children_.begin(); it != children_.end(); ++it) {
@@ -129,10 +132,10 @@
}
}
static std::map<ChildID, ForkedProcess *> children_;
- std::map<ChildID, volatile FunctionToCall<void> *> to_calls_;
+ std::map<ChildID, FunctionToCall<void> *> to_calls_;
- void SetUp() {
- SharedMemTestSetup::SetUp();
+ void SetUp() override {
+ ::testing::Test::SetUp();
fatal_failure = static_cast<char *>(shm_malloc(sizeof(fatal_failure)));
static bool registered = false;
if (!registered) {
@@ -145,24 +148,25 @@
// function gets called with arg in a forked process.
// Leaks shared memory.
template<typename T> __attribute__((warn_unused_result))
- std::unique_ptr<ForkedProcess> ForkExecute(void (*function)(T*), T *arg) {
+ std::unique_ptr<ForkedProcess> ForkExecute(void (*function)(T*), T *arg) {
mutex *lock = static_cast<mutex *>(shm_malloc_aligned(
sizeof(*lock), sizeof(int)));
- *lock = 1;
+ assert(mutex_lock(lock) == 0);
const pid_t pid = fork();
switch (pid) {
case 0: // child
- if (kForkSleep != 0) {
- printf("pid %jd sleeping for %u\n", static_cast<intmax_t>(getpid()),
- kForkSleep);
- sleep(kForkSleep);
+ if (kForkSleep != time::Time(0, 0)) {
+ LOG(INFO, "pid %jd sleeping for %ds%dns\n",
+ static_cast<intmax_t>(getpid()),
+ kForkSleep.sec(), kForkSleep.nsec());
+ time::SleepFor(kForkSleep);
}
- atexit(ExitExitHandler);
+ ::aos::common::testing::PreventExit();
function(arg);
mutex_unlock(lock);
exit(EXIT_SUCCESS);
case -1: // parent failure
- printf("fork() failed with %d: %s\n", errno, strerror(errno));
+ LOG(ERROR, "fork() failed with %d: %s\n", errno, strerror(errno));
return std::unique_ptr<ForkedProcess>();
default: // parent
return std::unique_ptr<ForkedProcess>(new ForkedProcess(pid, lock));
@@ -174,8 +178,8 @@
// NOTE: There are other reasons for it to return a failure than the function
// doing the wrong thing.
// Leaks shared memory.
- template<typename T> AssertionResult Hangs(void (*function)(T*, char*), T *arg,
- bool expected) {
+ template<typename T>
+ AssertionResult Hangs(void (*function)(T*, char*), T *arg, bool expected) {
AssertionResult fork_result(HangsFork<T>(function, arg, expected, 0));
if (!fork_result) {
return fork_result;
@@ -186,21 +190,24 @@
// Use HangsCheck to get the result.
// Returns whether the fork succeeded or not, NOT whether or not the hang
// check succeeded.
- template<typename T> AssertionResult HangsFork(void (*function)(T*, char *), T *arg,
- bool expected, ChildID id) {
+ template<typename T>
+ AssertionResult HangsFork(void (*function)(T*, char *), T *arg,
+ bool expected, ChildID id) {
static_assert(aos::shm_ok<FunctionToCall<T>>::value,
"this is going into shared memory");
- volatile FunctionToCall<T> *const to_call = static_cast<FunctionToCall<T> *>(
- shm_malloc_aligned(sizeof(*to_call), sizeof(int)));
- to_call->result = ResultType::NotCalled;
+ FunctionToCall<T> *const to_call =
+ static_cast<FunctionToCall<T> *>(
+ shm_malloc_aligned(sizeof(*to_call), alignof(FunctionToCall<T>)));
+ new (to_call) FunctionToCall<T>();
to_call->function = function;
to_call->arg = arg;
to_call->expected = expected;
to_call->failure[0] = '\0';
- static_cast<volatile char *>(fatal_failure)[0] = '\0';
+ static_cast<char *>(fatal_failure)[0] = '\0';
children_[id] = ForkExecute(Hangs_, to_call).release();
if (!children_[id]) return AssertionFailure() << "ForkExecute failed";
- to_calls_[id] = reinterpret_cast<volatile FunctionToCall<void> *>(to_call);
+ to_calls_[id] = reinterpret_cast<FunctionToCall<void> *>(to_call);
+ to_call->started.Lock();
return AssertionSuccess();
}
// Checks whether or not a function hung like it was supposed to.
@@ -222,8 +229,12 @@
<< ResultTypeString(to_calls_[id]->result));
} else {
if (to_calls_[id]->result == ResultType::Called) {
- return to_calls_[id]->expected ? AssertionSuccess() : AssertionFailure();
+ return to_calls_[id]->expected ? AssertionSuccess() :
+ AssertionFailure();
+ } else if (result == ForkedProcess::JoinResult::Error) {
+ return AssertionFailure() << "error joining child";
} else {
+ abort();
return AssertionFailure() << "something weird happened";
}
}
@@ -242,7 +253,8 @@
} while (false)
struct TestMessage {
- int16_t data; // don't really want to test empty messages
+ // Some contents because we don't really want to test empty messages.
+ int16_t data;
};
struct MessageArgs {
RawQueue *const queue;
@@ -252,7 +264,8 @@
static void WriteTestMessage(MessageArgs *args, char *failure) {
TestMessage *msg = static_cast<TestMessage *>(args->queue->GetMessage());
if (msg == NULL) {
- snprintf(fatal_failure, kFailureSize, "couldn't get_msg from %p", args->queue);
+ snprintf(fatal_failure, kFailureSize,
+ "couldn't get_msg from %p", args->queue);
return;
}
msg->data = args->data;
@@ -279,9 +292,14 @@
args->queue->FreeMessage(msg);
}
}
+
+ private:
+ GlobalCoreInstance my_core;
};
char *QueueTest::fatal_failure;
std::map<QueueTest::ChildID, QueueTest::ForkedProcess *> QueueTest::children_;
+constexpr time::Time QueueTest::kHangTime;
+constexpr time::Time QueueTest::kForkSleep;
TEST_F(QueueTest, Reading) {
RawQueue *const queue = RawQueue::Fetch("Queue", sizeof(TestMessage), 1, 1);
@@ -363,8 +381,8 @@
// TODO(brians) basic test of recycle queue
// include all of the ways a message can get into the recycle queue
RawQueue *recycle_queue = reinterpret_cast<RawQueue *>(23);
- RawQueue *const queue = RawQueue::Fetch("Queue", sizeof(TestMessage), 1, 2, 2, 2,
- &recycle_queue);
+ RawQueue *const queue = RawQueue::Fetch("Queue", sizeof(TestMessage),
+ 1, 2, 2, 2, &recycle_queue);
ASSERT_NE(reinterpret_cast<RawQueue *>(23), recycle_queue);
MessageArgs args{queue, 0, 973}, recycle{recycle_queue, 0, 973};
diff --git a/aos/atom_code/ipc_lib/sharedmem_test_setup.h b/aos/atom_code/ipc_lib/sharedmem_test_setup.h
deleted file mode 100644
index e461c43..0000000
--- a/aos/atom_code/ipc_lib/sharedmem_test_setup.h
+++ /dev/null
@@ -1,147 +0,0 @@
-// defines a fixture (SharedMemTestSetup) that sets up shared memory
-
-extern "C" {
-#include "shared_mem.h"
- extern struct aos_core *global_core;
-}
-
-#include <signal.h>
-
-#include <gtest/gtest.h>
-#include <sys/types.h>
-
-// TODO(brians) read logs from here
-class SharedMemTestSetup : public testing::Test{
- protected:
- pid_t core;
- int start[2];
- int memcheck[2];
- static void signal_handler(int){
- if(aos_core_free_shared_mem()){
- exit(- 1);
- }
- exit(0);
- }
- static int get_mem_usage(){
- return global_core->size - ((uint8_t *)global_core->mem_struct->msg_alloc - (uint8_t *)SHM_START);
- }
- bool checking_mem;
-
- virtual void BeforeLocalShmSetup() {}
- virtual void SetUp(){
- ASSERT_EQ(0, pipe(start)) << "couldn't create start pipes";
- ASSERT_EQ(0, pipe(memcheck)) << "couldn't create memcheck pipes";
- checking_mem = false;
- if((core = fork()) == 0){
- close(start[0]);
- close(memcheck[1]);
- struct sigaction act;
- act.sa_handler = signal_handler;
- sigaction(SIGINT, &act, NULL);
- if(aos_core_create_shared_mem(create)){
- exit(-1);
- }
- write_pipe(start[1], "a", 1);
- int usage = 0;
- while(1){
- char buf1;
- read_pipe(memcheck[0], &buf1, 1);
- if(usage == 0)
- usage = get_mem_usage();
- if(usage == get_mem_usage())
- buf1 = 1;
- else
- buf1 = 0;
- write_pipe(start[1], &buf1, 1);
- }
- }
- close(start[1]);
- close(memcheck[0]);
- ASSERT_NE(-1, core) << "fork failed";
- char buf;
- read_pipe(start[0], &buf, 1);
-
- BeforeLocalShmSetup();
-
- ASSERT_EQ(0, aos_core_create_shared_mem(reference)) << "couldn't create shared mem reference";
- }
- virtual void TearDown(){
- if(checking_mem){
- write_pipe(memcheck[1], "a", 1);
- char buf;
- read_pipe(start[0], &buf, 1);
- EXPECT_EQ(1, buf) << "memory got leaked";
- }
- EXPECT_EQ(0, aos_core_free_shared_mem()) << "issues freeing shared mem";
- if(core > 0){
- kill(core, SIGINT);
- siginfo_t status;
- ASSERT_EQ(0, waitid(P_PID, core, &status, WEXITED)) << "waiting for the core to finish failed";
- EXPECT_EQ(CLD_EXITED, status.si_code) << "core died";
- EXPECT_EQ(0, status.si_status) << "core exited with an error";
- }
- }
- // if any more shared memory gets allocated after calling this and not freed by the end, it's an error
- void AllSharedMemAllocated(){
- checking_mem = true;
- write_pipe(memcheck[1], "a", 1);
- char buf;
- read_pipe(start[0], &buf, 1);
- }
- private:
- // Wrapper functions for pipes because they should never have errors.
- void read_pipe(int fd, void *buf, size_t count) {
- if (read(fd, buf, count) < 0) abort();
- }
- void write_pipe(int fd, const void *buf, size_t count) {
- if (write(fd, buf, count) < 0) abort();
- }
-};
-class ExecVeTestSetup : public SharedMemTestSetup {
- protected:
- std::vector<std::string> files;
- std::vector<pid_t> pids;
- virtual void BeforeLocalShmSetup(){
- std::vector<std::string>::iterator it;
- pid_t child;
- for(it = files.begin(); it < files.end(); ++it){
- if((child = fork()) == 0){
- char *null = NULL;
- execve(it->c_str(), &null, &null);
- ADD_FAILURE() << "execve failed";
- perror("execve");
- exit(0);
- }
- if(child > 0)
- pids.push_back(child);
- else
- ADD_FAILURE() << "fork failed return=" << child;
- }
- usleep(150000);
- }
- virtual void TearDown(){
- std::vector<pid_t>::iterator it;
- siginfo_t status;
- for(it = pids.begin(); it < pids.end(); ++it){
- printf("attempting to SIGINT(%d) %d\n", SIGINT, *it);
- if(*it > 0){
- kill(*it, SIGINT);
- ASSERT_EQ(0, waitid(P_PID, *it, &status, WEXITED)) << "waiting for the AsyncAction(pid=" << *it << ") to finish failed";
- EXPECT_EQ(CLD_EXITED, status.si_code) << "child died (killed by signal is " << (int)CLD_KILLED << ")";
- EXPECT_EQ(0, status.si_status) << "child exited with an error";
- }else{
- FAIL();
- }
- }
-
- SharedMemTestSetup::TearDown();
- }
- // call this _before_ ExecVeTestSetup::SetUp()
- void AddProcess(const std::string file){
- files.push_back(file);
- }
- void PercolatePause(){
- usleep(50000);
- }
-};
-
diff --git a/aos/build/aos_all.gyp b/aos/build/aos_all.gyp
index de15572..e4a9b6f 100644
--- a/aos/build/aos_all.gyp
+++ b/aos/build/aos_all.gyp
@@ -13,7 +13,7 @@
'../atom_code/camera/camera.gyp:CameraHTTPStreamer',
'../atom_code/camera/camera.gyp:CameraReader',
'../atom_code/core/core.gyp:*',
- '../atom_code/ipc_lib/ipc_lib.gyp:ipc_queue_test',
+ '../atom_code/ipc_lib/ipc_lib.gyp:raw_queue_test',
'../atom_code/ipc_lib/ipc_lib.gyp:ipc_stress_test',
'../atom_code/starter/starter.gyp:starter_exe',
'../atom_code/starter/starter.gyp:netconsole',
diff --git a/aos/common/common.gyp b/aos/common/common.gyp
index eee1d05..f3b4e69 100644
--- a/aos/common/common.gyp
+++ b/aos/common/common.gyp
@@ -24,7 +24,11 @@
'<(AOS)/build/aos.gyp:logging',
'once',
'<(EXTERNALS):gtest',
+ '<(AOS)/atom_code/ipc_lib/ipc_lib.gyp:shared_mem',
],
+ 'export_dependent_settings': [
+ '<(AOS)/atom_code/ipc_lib/ipc_lib.gyp:shared_mem',
+ ],
},
{
'target_name': 'time',
diff --git a/aos/common/condition.h b/aos/common/condition.h
index 346ae54..c407070 100644
--- a/aos/common/condition.h
+++ b/aos/common/condition.h
@@ -7,8 +7,38 @@
namespace aos {
// A condition variable (IPC mechanism where 1 process/task can notify all
-// others that are waiting for something to happen).
-// This implementation will LOG(FATAL) if anything weird happens.
+// others that are waiting for something to happen) without the race condition
+// where a notification is sent after some process has checked if the thing has
+// happened but before it has started listening for notifications.
+//
+// This implementation will print debugging information and abort the process
+// if anything weird happens.
+//
+// A simple example of the use of a condition variable (adapted from
+// pthread_cond(3)):
+//
+// int x, y;
+// Mutex mutex;
+// Condition condition(&mutex);
+//
+// // Waiting until x is greater than y:
+// {
+// MutexLocker locker(&mutex);
+// while (!(x > y)) condition.Wait();
+// // do whatever
+// }
+//
+// // Modifying x and/or y:
+// {
+// MutexLocker locker(&mutex);
+// // modify x and y
+// if (x > y) condition.Broadcast();
+// }
+//
+// Notice the loop around the Wait(). This is very important because some other
+// process can lock the mutex and modify the shared state (possibly undoing
+// whatever the Wait()er was waiting for) in between the Broadcast()er unlocking
+// the mutex and the Wait()er(s) relocking it.
//
// Multiple condition variables may be associated with the same mutex but
// exactly 1 mutex must be associated with each condition variable.
@@ -18,9 +48,12 @@
// object will hold on to a reference to it but does not take ownership.
explicit Condition(Mutex *m);
- // Waits for the condition variable to be signalled, atomically unlocking m at
- // the same time. The mutex associated with this condition variable must be
- // locked when this is called and will be locked when this method returns.
+ // Waits for the condition variable to be signalled, atomically unlocking the
+ // mutex associated with this condition variable at the same time. The mutex
+ // associated with this condition variable must be locked when this is called
+ // and will be locked when this method returns.
+ // NOTE: The relocking of the mutex is not performed atomically with waking
+ // up.
void Wait();
// Signals at most 1 other process currently Wait()ing on this condition
diff --git a/aos/common/condition_test.cc b/aos/common/condition_test.cc
index 49112bc..28b4adc 100644
--- a/aos/common/condition_test.cc
+++ b/aos/common/condition_test.cc
@@ -76,6 +76,7 @@
child_ = fork();
if (child_ == 0) { // in child
+ ::aos::common::testing::PreventExit();
Run();
exit(EXIT_SUCCESS);
} else { // in parent
@@ -131,11 +132,11 @@
ready.Lock();
}
- bool started;
- bool delayed;
+ volatile bool started;
+ volatile bool delayed;
Mutex done_delaying;
::Time start_time;
- bool finished;
+ volatile bool finished;
Mutex ready;
};
static_assert(shm_ok<Shared>::value,
diff --git a/aos/common/mutex_test.cpp b/aos/common/mutex_test.cpp
index 6ab7ed6..652cd9e 100644
--- a/aos/common/mutex_test.cpp
+++ b/aos/common/mutex_test.cpp
@@ -72,105 +72,5 @@
EXPECT_FALSE(test_mutex.TryLock());
}
-// A worker thread for testing the fairness of the mutex implementation.
-class MutexFairnessWorkerThread {
- public:
- MutexFairnessWorkerThread(int *cycles, int index,
- Mutex *in_mutex, mutex *start)
- : cycles_(cycles), index_(index), mutex_(in_mutex), start_(start) {}
-
- static void *RunStatic(void *self_in) {
- MutexFairnessWorkerThread *self =
- static_cast<MutexFairnessWorkerThread *>(self_in);
- self->Run();
- delete self;
- return NULL;
- }
-
- static void Reset(int cycles) {
- cyclesRun = 0;
- totalCycles = cycles;
- }
-
- private:
- void Run() {
- cycles_[index_] = 0;
- ASSERT_EQ(futex_wait(start_), 0);
- while (cyclesRun < totalCycles) {
- {
- MutexLocker locker(mutex_);
- ++cyclesRun;
- }
- ++cycles_[index_];
- // Otherwise the fitpc implementation tends to just relock in the same
- // thread.
- sched_yield();
- }
-
-#ifdef __VXWORKS__
- // Without this, all of the "task ... deleted ..." messages come out at
- // once, and it looks weird and triggers an socat bug (at least for
- // Squeeze's version 1.7.1.3-1) which locks it up and is very annoying.
- taskDelay(index_);
-#endif
- }
-
- int *cycles_;
- int index_;
- Mutex *mutex_;
- mutex *start_;
- static int cyclesRun, totalCycles;
-};
-int MutexFairnessWorkerThread::cyclesRun;
-int MutexFairnessWorkerThread::totalCycles;
-
-// Tests the fairness of the implementation. It does this by repeatedly locking
-// and unlocking a mutex in multiple threads and then checking the standard
-// deviation of the number of times each one locks.
-//
-// It is safe to do this with threads because this is the test so it can change
-// if the implementations ever change to not support that. Fitpc logging calls
-// are not thread-safe, but it doesn't really matter because the only logging
-// call that would get made would be a LOG(FATAL) that would still terminate the
-// process.
-TEST_F(MutexTest, Fairness) {
- static const int kThreads = 13;
-#ifdef __VXWORKS__
- static const int kWarmupCycles = 1000, kRunCycles = 60000, kMaxDeviation = 20;
-#else
- static const int kWarmupCycles = 30000, kRunCycles = 3000000, kMaxDeviation = 20000;
-#endif
-
- int cycles[kThreads];
- pthread_t workers[kThreads];
- mutex start = 0;
-
- for (int repeats = 0; repeats < 2; ++repeats) {
- futex_unset(&start);
- MutexFairnessWorkerThread::Reset(repeats ? kRunCycles : kWarmupCycles);
- for (int i = 0; i < kThreads; ++i) {
- MutexFairnessWorkerThread *c = new MutexFairnessWorkerThread(cycles, i,
- &test_mutex,
- &start);
- ASSERT_EQ(0, pthread_create(&workers[i], NULL,
- MutexFairnessWorkerThread::RunStatic, c));
- }
- futex_set(&start);
- for (int i = 0; i < kThreads; ++i) {
- ASSERT_EQ(0, pthread_join(workers[i], NULL));
- }
- }
-
- double variance = 0;
- int expected = kRunCycles / kThreads;
- for (int i = 0; i < kThreads; ++i) {
- variance += (cycles[i] - expected) * (cycles[i] - expected);
- }
- ASSERT_GT(variance, 0);
- double deviation = sqrt(variance / kThreads);
- ASSERT_GT(deviation, 0);
- EXPECT_LT(deviation, kMaxDeviation);
-}
-
} // namespace testing
} // namespace aos
diff --git a/aos/common/queue_testutils.cc b/aos/common/queue_testutils.cc
index 629c9be..8d195c5 100644
--- a/aos/common/queue_testutils.cc
+++ b/aos/common/queue_testutils.cc
@@ -2,6 +2,9 @@
#include <string.h>
#include <sys/mman.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <assert.h>
#include "gtest/gtest.h"
@@ -113,13 +116,18 @@
const size_t kCoreSize = 0x100000;
+void TerminateExitHandler() {
+ _exit(EXIT_SUCCESS);
+}
+
} // namespace
GlobalCoreInstance::GlobalCoreInstance() {
global_core = &global_core_data_;
- global_core->owner = 1;
- // Use mmap(2) manually so that we can pass MAP_SHARED so that forked
- // processes can still communicate using the "shared" memory.
+ global_core->owner = true;
+ // Use mmap(2) manually instead of through malloc(3) so that we can pass
+ // MAP_SHARED which allows forked processes to communicate using the
+ // "shared" memory.
void *memory = mmap(NULL, kCoreSize, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
assert(memory != MAP_FAILED);
@@ -138,6 +146,10 @@
enable_test_logging_once.Get();
}
+void PreventExit() {
+ assert(atexit(TerminateExitHandler) == 0);
+}
+
} // namespace testing
} // namespace common
} // namespace aos
diff --git a/aos/common/queue_testutils.h b/aos/common/queue_testutils.h
index aabdd2d..2d26262 100644
--- a/aos/common/queue_testutils.h
+++ b/aos/common/queue_testutils.h
@@ -1,11 +1,20 @@
-#include "aos/common/queue.h"
+#ifndef AOS_COMMON_QUEUE_TESTUTILS_H_
+#define AOS_COMMON_QUEUE_TESTUTILS_H_
+
+#include "aos/atom_code/ipc_lib/shared_mem.h"
+
+// This file has some general helper functions for dealing with testing things
+// that use shared memory etc.
namespace aos {
namespace common {
namespace testing {
+// Manages creating and cleaning up "shared memory" which works within this
+// process and any that it fork(2)s.
class GlobalCoreInstance {
public:
+ // Calls EnableTestLogging().
GlobalCoreInstance();
~GlobalCoreInstance();
@@ -20,6 +29,14 @@
// initialized), however it can be called more than that.
void EnableTestLogging();
+// Registers an exit handler (using atexit(3)) which will call _exit(2).
+// Intended to be called in a freshly fork(2)ed process where it will run before
+// any other exit handlers that were already registered and prevent them from
+// being run.
+void PreventExit();
+
} // namespace testing
} // namespace common
} // namespace aos
+
+#endif // AOS_COMMON_QUEUE_TESTUTILS_H_