wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enable IThreadPool in simulation

This commit is contained in:
Daniel Smith 2021-07-08 18:51:01 -04:00
parent 0452ff5759
commit 3fbd6b6143
6 changed files with 16 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
fdbrpc/FlowTests.actor.cpp
View File

@ -262,8 +262,8 @@ struct YieldMockNetwork final : INetwork, ReferenceCounted<YieldMockNetwork> {
return baseNetwork->onMainThread(std::move(signal), taskID);
}
bool isOnMainThread() const override { return baseNetwork->isOnMainThread(); }
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg) override {
return baseNetwork->startThread(func, arg);
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg, int stackSize, const char* name) override {
return baseNetwork->startThread(func, arg, stackSize, name);
}
Future<Reference<class IAsyncFile>> open(std::string filename, int64_t flags, int64_t mode) {
return IAsyncFileSystem::filesystem()->open(filename, flags, mode);

4
fdbrpc/sim2.actor.cpp
View File

@ -1007,9 +1007,9 @@ public:
THREAD_RETURN;
}
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg) override {
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg, int stackSize, const char* name) override {
SimThreadArgs* simArgs = new SimThreadArgs(func, arg);
return ::startThread(simStartThread, simArgs);
return ::startThread(simStartThread, simArgs, stackSize, name);
}
void getDiskBytes(std::string const& directory, int64_t& free, int64_t& total) override {

2
flow/IThreadPool.cpp
View File

@ -109,7 +109,7 @@ public:
}
void addThread(IThreadPoolReceiver* userData, const char* name) override {
threads.push_back(new Thread(this, userData));
threads.back()->handle = startThread(start, threads.back(), stackSize, name);
threads.back()->handle = g_network->startThread(start, threads.back(), stackSize, name);
}
void post(PThreadAction action) override { ios.post(ActionWrapper(action)); }
};

2
flow/IThreadPoolTest.actor.cpp
View File

@ -35,7 +35,7 @@ struct ThreadNameReceiver : IThreadPoolReceiver {
}
};
TEST_CASE("noSim/IThreadPool/NamedThread") {
TEST_CASE("/flow/IThreadPool/NamedThread") {
state Reference<IThreadPool> pool = createGenericThreadPool();
pool->addThread(new ThreadNameReceiver(), "thread-foo");

8
flow/Net2.actor.cpp
View File

@ -184,7 +184,7 @@ public:
}
bool isSimulated() const override { return false; }
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg) override;
THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg, int stackSize, const char* name) override;
void getDiskBytes(std::string const& directory, int64_t& free, int64_t& total) override;
bool isAddressOnThisHost(NetworkAddress const& addr) const override;
@ -1513,7 +1513,7 @@ void Net2::run() {
double newTaskBegin = timer_monotonic();
if (check_yield(TaskPriority::Max, tscNow)) {
checkForSlowTask(tscBegin, tscNow, newTaskBegin - taskBegin, currentTaskID);
taskBegin = newTaskBegin;
taskBegin = newTaskBegin;
FDB_TRACE_PROBE(run_loop_yield);
++countYields;
break;
@ -1765,8 +1765,8 @@ void Net2::onMainThread(Promise<Void>&& signal, TaskPriority taskID) {
}
}
THREAD_HANDLE Net2::startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg) {
return ::startThread(func, arg);
THREAD_HANDLE Net2::startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg, int stackSize, const char* name) {
return ::startThread(func, arg, stackSize, name);
}
Future<Reference<IConnection>> Net2::connect(NetworkAddress toAddr, const std::string& host) {

8
flow/network.h
View File

@ -347,7 +347,8 @@ struct NetworkMetrics {
std::unordered_map<TaskPriority, struct PriorityStats> activeTrackers;
double lastRunLoopBusyness; // network thread busyness (measured every 5s by default)
std::atomic<double> networkBusyness; // network thread busyness which is returned to the the client (measured every 1s by default)
std::atomic<double>
networkBusyness; // network thread busyness which is returned to the the client (measured every 1s by default)
// starvation trackers which keeps track of different task priorities
std::vector<struct PriorityStats> starvationTrackers;
@ -536,7 +537,10 @@ public:
virtual void onMainThread(Promise<Void>&& signal, TaskPriority taskID) = 0;
// Executes signal.send(Void()) on a/the thread belonging to this network
virtual THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*), void* arg) = 0;
virtual THREAD_HANDLE startThread(THREAD_FUNC_RETURN (*func)(void*),
void* arg,
int stackSize = 0,
const char* name = nullptr) = 0;
// Starts a thread and returns a handle to it
virtual void run() = 0;