Multi Threading Library is a set of utilities, easy and ready to use for common task in multi threading environment and development.
- Thread Class
- Thread Manager
- Thread Pool
- Task Class
- Ordered Task
- Task Flow
- Complete Documentation
- Doxygen Integration
- First Beta Release
- Test Framework
- CI/CD Pipeline
- First Stable Release
- Thread Monitoring
- ...
See the open issues for a full list of proposed features (and known issues).
This is an example of how you may give instructions on setting up your project locally. To get a local copy up and running follow these simple example steps.
The following are required to get the project up and running.
git clone https://github.com/google/googletest.git # Dowload the Google Test repository
cd googletest # Main directory of the cloned repository.
mkdir -p build # Create a directory to hold the build output.
cd build # Move into the build directory.
cmake .. # Generate native build scripts for GoogleTest.
make # Compile
sudo make install # Install in /usr/local/ by defaultgit clone https://github.com/ZigRazor/MTL.git # Dowload the MTL repository
cd MTL # Main directory of the cloned repository.
mkdir -p build # Create a directory to hold the build output.
cd build # Move into the build directory.
cmake .. # Generate native build scripts for MTL.
make # Compile
sudo make install # Install in /usr/local/ by defaultThe only things you have to do to get library in you project are to:
- Include the header file MTL.h in your project.
- Link the library MTL.so to your project.
To let the thread class know what to do, you must create a Runnable object.
MyRunnable.hpp
#include <iostream>
#include "MTL.h"
class MyRunnable : public MTL::MTLRunnable
{
public:
MyRunnable() = default;
virtual ~MyRunnable() = default;
void run(MTL::MTLThreadInterface* threadIf)
{
std::cout << "Hello World!" << std::endl;
int counter = 0;
while (true)
{
if(threadIf->getThreadState() == MTL::E_MTLThreadState::STOPPED){
std::cout << "Stopped" << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}else if (threadIf->getThreadState() == MTL::E_MTLThreadState::SUSPENDED)
{
std::cout << "Suspended" << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}else if (threadIf->getThreadState() == MTL::E_MTLThreadState::EXITED)
{
std::cout << "Exited" << std::endl;
break;
} else if (threadIf->getThreadState() == MTL::E_MTLThreadState::RUNNING)
{
//Simulate 20 seconds of running work
if(counter == 20)
{
threadIf->setThreadState(MTL::E_MTLThreadState::EXITED);
}
std::cout << "Running" << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
counter++;
}
}
}
void stop()
{
std::cout << "Stop!" << std::endl;
}
void suspend()
{
std::cout << "Pause!" << std::endl;
}
void resume()
{
std::cout << "Resume!" << std::endl;
}
void clean_exit()
{
std::cout << "Clean Exit!" << std::endl;
}
void force_exit()
{
std::cout << "Force Exit!" << std::endl;
::exit(0); // clean Exit
}
};Then in the main function you can create a thread object and pass it the runnable object.
main.cpp
#include "MyRunnable.hpp"
int main()
{
MyRunnable myRunnable; // Create a runnable object
MTL::MTLThread thread(myRunnable); // Create a thread object and pass it the runnable object.
thread.run(); //Start the Thread
std::this_thread::sleep_for(std::chrono::milliseconds(10000)); //Sleep for 10 seconds
thread.suspend(); //Suspend the thread
std::this_thread::sleep_for(std::chrono::milliseconds(5000)); //Sleep for 5 seconds
thread.resume(); //Resume the thread
int counter = 0;
while(thread.isRunning()){ //Wait 10 seconds in the main thread
if(counter == 10)
{
break;
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Counter: " << counter << std::endl;
counter++;
}
return 0;
}The Worker Thread class is a runnable with 2 more features: - It contains a message queue - Has to implement the function to process the messages
This class can be used to create a runnable to pass to a Thread object that receive an input message and elaborate it.
A simple example of implementation of the Worker Thread Class can be the following:
MyWorkerThread.hpp
class MyWorker : public MTL::MTLWorkerThread
{
public:
MyWorker() = default;
virtual ~MyWorker() = default;
virtual void processMessage(MTL::Message message) override
{
//When the message is dequeued it is passed to this function
int* message_casted = static_cast< int *> (message.get()); //Cast the message to the correct type
std::cout << "MyWorker::processMessage(" << *message_casted << ")" << std::endl; //Do somenthings with the message
}
};The Thread Manager class is a Runnable that can be used to manage threads. If you want to manage a group of threads all togheter you can create a Thread Manager object and pass it to a Thread object. The threads can be passed to the Thread Manager object and the Thread Manager object will manage the threads. So if you have multiple tasks or multiple workers, you can collect them in a thread manager and all the action performed to the thread manager(start, stop, suspend, resume, etc) will be performed to all the threads. A Simple example of usage can be the following:
#include "MTL.h"
#include "MyWorker.hpp"
int main(){
// Three different kind of workers
MyWorker1 myWorker1;
MyWorker2 myWorker2;
MyWorker3 myWorker3;
MTL::MTLThreadManager threadManager; //Create a thread manager object
threadManager.addThread(std::make_unique<MTL::MTLThread>(myWorker1)); //Add a thread to the thread manager
threadManager.addThread(std::make_unique<MTL::MTLThread>(myWorker2)); //Add a thread to the thread manager
MTL::MTLThread thread(threadManager); //Create a thread object and pass it the thread manager
thread.run(); //Start the thread manager that will start all the threads
std::this_thread::sleep_for(std::chrono::milliseconds(5000)); //Let him work for 5 seconds
thread.suspend(); //Suspend the thread manager that will suspend all the threads
std::this_thread::sleep_for(std::chrono::milliseconds(5000)); //Let suspended for 5 seconds
thread.resume(); //Resume the thread manager that will resume all the threads
threadManager.addThread(std::make_unique<MTL::MTLThread>(myWorker3)); //Add a thread to the thread manager
thread.clean_exit(); // Clean Exit the thread manager that will clean exit all the threads
while (thread.isRunning()) //Wait for the thread manager to exit
{
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Counter: " << counter << std::endl;
counter++;
}
std::cout << "Thread Manager Terminated" << std::endl;
return 0;
}The Thread Pool class is a Thread Manager that can be used to create a fixed number of workers that execute the same task. When a message is enqueued to the thread pool, the message is passed to a workers. This kind of thread manager is useful when you have to implement a multi threaded consumer. An example of usage can be the following:
#include "MTL.h"
int main(){
MyWorker1 myWorker1; //Create a worker
MTL::MTLThreadPool threadPool(myWorker1, 4); //Create a thread pool with 4 workers
MTL::MTLThread thread(threadPool); //Create a thread object and pass it the thread pool
thread.run(); //Start the thread pool that will start all the workers
for (int i = 0; i < 10; ++i) // Enqueue 10 messages
{
std::cout << "Inject Message " << i << std::endl;
MTL::Message message(new int(i));
threadPool.onMessage(message);
}
std::this_thread::sleep_for(std::chrono::milliseconds(10000)); //Let him work for 10 seconds
std::cout << "Suspend" << std::endl;
thread.suspend(); // Suspend the thread pool that will suspend all the workers
for (int i = 10; i < 20; ++i) // Enqueue 10 messages
{
std::cout << "Inject Message " << i << std::endl;
MTL::Message message(new int(i));
threadPool.onMessage(message);
}
std::this_thread::sleep_for(std::chrono::milliseconds(5000)); //Let Suspeded for 5 seconds
std::cout << "Resume" << std::endl;
thread.resume(); // Resume the thread pool that will resume all the workers
std::cout << "Exit" << std::endl;
thread.clean_exit(); // Clean Exit the thread pool that will clean exit all the workers
while (thread.isRunning()) //Wait for the thread pool to exit
{
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Counter: " << counter << std::endl;
counter++;
}
std::cout << "Thread Pool Terminated" << std::endl;
return 0;
}The Shared Object class is a class that manage a shared memory object and can be used in a thread safe manner.
The Shared Memory class is a class that manage shared objects and can be used in a thread safe manner.
The two classes can be used to create a thread safe memory to share data between threads.
The following is a simple example of application:
MySharedObject.hpp
#include "MTL.h"
class MySharedObject : public MTL::MTLSharedObject {
public:
MySharedObject(unsigned int id) : MTLSharedObject(id) {
std::cout << "MySharedObject::MySharedObject()" << std::endl;
value = 0;
}
~MySharedObject() {
std::cout << "MySharedObject::~MySharedObject()" << std::endl;
}
int getValue() {
return value;
}
void setValue(int v) {
value = v;
}
private:
int value; //The value of the shared object
};MyRunnable.hpp
#include "MTL.h"
class MyRunnable : public MTL::MTLRunnable
{
public:
MyRunnable(MTL::MTLSharedMemory* sharedMemory) : m_sharedMemory(sharedMemory)
{};
virtual ~MyRunnable() = default;
void run(MTL::MTLThreadInterface* threadIf)
{
std::cout << "Hello World!" << std::endl;
int counter = 0;
while (true)
{
if(threadIf->getThreadState() == MTL::E_MTLThreadState::STOPPED){
std::cout << "Stopped Thread Id: " << std::this_thread::get_id() << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}else if (threadIf->getThreadState() == MTL::E_MTLThreadState::SUSPENDED)
{
std::cout << "Suspended Thread Id: "<< std::this_thread::get_id() << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}else if (threadIf->getThreadState() == MTL::E_MTLThreadState::EXITED)
{
std::cout << "Exited Thread Id: " << std::this_thread::get_id() << std::endl;
break;
} else if (threadIf->getThreadState() == MTL::E_MTLThreadState::RUNNING)
{
// when running this runnable increment the shared memory value by 1
MySharedObject& myObj = dynamic_cast<MySharedObject&>(m_sharedMemory->getSharedObjectById(1));
std::cout << "Thread Id: " << std::this_thread::get_id() << " starting Value: " << myObj.getValue() << " end Value: " << myObj.getValue() + 1 << std::endl;
myObj.setValue(myObj.getValue() + 1);
m_sharedMemory->releaseSharedObject(myObj);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
counter++;
}
}
}
void stop()
{
std::cout << "Stopping" << std::endl;
}
void suspend()
{
std::cout << "Suspending" << std::endl;
}
void resume()
{
std::cout << "Resuming" << std::endl;
}
void clean_exit()
{
std::cout << "Exiting" << std::endl;
}
void force_exit()
{
std::cout << "Force Exiting" << std::endl;
}
private:
MTL::MTLSharedMemory* m_sharedMemory;
};main.cpp
#include "MTL.h"
#include "MySharedObject.hpp"
#include "MyRunnable.hpp"
int main()
{
/**
* This example demostrate how the shared memory and shared object classes
* can be used to share data between threads. Without alter the atomicity
* of the execution over the memory objects
**/
std::unique_ptr<MySharedObject> myObj(new MySharedObject(1)); //Create a shared object
MTL::MTLSharedMemory sharedMemory; //Create a shared memory
sharedMemory.addSharedObject(std::move(myObj)); //Add the shared object to the shared memory
MyRunnable myRunnable1(&sharedMemory); //Create a runnable
MyRunnable myRunnable2(&sharedMemory); //Create another runnable
MTL::MTLThread thread1(myRunnable1); //Create a thread from the runnable
MTL::MTLThread thread2(myRunnable2); //Create another thread from the runnable
thread1.run(); //Run the thread
thread2.run(); //Run the thread
std::this_thread::sleep_for(std::chrono::milliseconds(1000)); //Let the threads run for 1 second
std::cout << "Suspend Thread 1" << std::endl;
thread1.suspend(); //Suspend the thread 1
std::this_thread::sleep_for(std::chrono::milliseconds(1000)); //Let the thread 2 runs for 1 second
std::cout << "Resume Thread 1" << std::endl;
thread1.resume(); //Resume the thread 1
std::this_thread::sleep_for(std::chrono::milliseconds(1000)); //Let the threads run for 1 second
std::cout << "Exit the Threads" << std::endl;
thread1.clean_exit(); //Exit the thread 1
thread2.clean_exit(); //Exit the thread 2
thread1.join(); //Wait for the thread 1 to finish
thread2.join(); //Wait for the thread 2 to finish
std::cout << "Threads Joined" << std::endl;
return 0;
}The task class is a wrapper of C++ task class. It needs a runnable task to be executed. A simple example can be the following one:
MyRunnableTask.hpp
#include <iostream>
#include "MTL.h"
class MyRunnableTask : public MTL::MTLRunnableTask
{
public:
MyRunnableTask() = default;
virtual ~MyRunnableTask() = default;
std::shared_ptr<void> run(MTL::MTLTaskInterface *interface = nullptr)
{
std::cout << "Hello World!" << std::endl;
std::cout << "Simulating Working for 3 seconds" << std::endl;
for (int i = 0; i < 3; i++)
{
std::cout << "." << std::flush;
std::this_thread::sleep_for(std::chrono::seconds(1));
}
std::cout << std::endl;
std::cout << "This is a Task that return 1" << std::endl;
int i = 1;
std::shared_ptr<void> result(new int(i));
return result;
}
};main.cpp
#include "MyRunnableTask.hpp"
int main()
{
MyRunnableTask myRunnableTask; //Create a runnable task
MTL::MTLTask task(myRunnableTask); //Create a task from the runnable task
task.run(); //Run the task
std::shared_ptr<void> result = task.getResult(); //Get the result of the task
std::cout << "Result: " << *(static_cast<int *>(result.get())) << std::endl; //Print the result
return 0;
}The Ordered Task class is derived from Task class and has a list of predecessors and a list of successors.
The Task Flow class is a class that allow to execute ordered task in a consistent manner and return the final result.
A simple example can be the following one:
DerivedTasks.hpp
#include <iostream>
#include "MTL.h"
//Return 2
class Var2Task : public MTL::MTLRunnableTask
{
public:
Var2Task() = default;
virtual ~Var2Task() = default;
std::shared_ptr<void> run(MTL::MTLTaskInterface *interface = nullptr)
{
MTL::MTLOrderedTaskInterface *orderedTaskIf = dynamic_cast<MTL::MTLOrderedTaskInterface *>(interface);
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Start Task " << std::endl
<< std::flush;
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Simulating Working for 3 seconds" << std::endl
<< std::flush;
for (int i = 0; i < 3; i++)
{
std::cout << "." << std::flush;
std::this_thread::sleep_for(std::chrono::seconds(1));
}
std::cout << std::endl;
auto previousResult = orderedTaskIf->getPredecessorsResults();
std::shared_ptr<void> result(new int(2));
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Result = " << *((int *)(result.get())) << std::endl
<< std::flush;
return result;
}
};
// Sum predecessors
class SumTask : public MTL::MTLRunnableTask
{
public:
SumTask() = default;
virtual ~SumTask() = default;
std::shared_ptr<void> run(MTL::MTLTaskInterface *interface = nullptr)
{
MTL::MTLOrderedTaskInterface *orderedTaskIf = dynamic_cast<MTL::MTLOrderedTaskInterface *>(interface);
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Start Task " << std::endl
<< std::flush;
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Simulating Working for 3 seconds" << std::endl
<< std::flush;
for (int i = 0; i < 3; i++)
{
std::cout << "." << std::flush;
std::this_thread::sleep_for(std::chrono::seconds(1));
}
std::cout << std::endl;
auto predecessorsResults = orderedTaskIf->getPredecessorsResults();
std::shared_ptr<void> result(new int(0));
if (predecessorsResults.empty())
{
std::shared_ptr<void> result(new int(0));
return result;
}
else
{
auto it = predecessorsResults.begin();
std::shared_ptr<void> result((int *)it->second.get());
++it;
for (it; it != predecessorsResults.end(); ++it)
{
int *previousResult = (int *)(it->second.get());
int *currentResult = (int *)(result.get());
*currentResult = (*currentResult) + (*previousResult);
}
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Result = " << *((int *)(result.get())) << std::endl
<< std::flush;
return result;
}
}
};
// Multiply predecessors
class MulTask : public MTL::MTLRunnableTask
{
public:
MulTask() = default;
virtual ~MulTask() = default;
std::shared_ptr<void> run(MTL::MTLTaskInterface *interface = nullptr)
{
MTL::MTLOrderedTaskInterface *orderedTaskIf = dynamic_cast<MTL::MTLOrderedTaskInterface *>(interface);
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Start Task " << std::endl
<< std::flush;
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Simulating Working for 3 seconds" << std::endl
<< std::flush;
for (int i = 0; i < 3; i++)
{
std::cout << "." << std::flush;
std::this_thread::sleep_for(std::chrono::seconds(1));
}
std::cout << std::endl
<< std::flush;
auto predecessorsResults = orderedTaskIf->getPredecessorsResults();
std::shared_ptr<void> result(new int(0));
if (predecessorsResults.empty())
{
std::shared_ptr<void> result(new int(0));
return result;
}
else
{
auto it = predecessorsResults.begin();
std::shared_ptr<void> result((int *)it->second.get());
++it;
for (it; it != predecessorsResults.end(); ++it)
{
int *previousResult = (int *)(it->second.get());
int *currentResult = (int *)(result.get());
*currentResult = (*currentResult) * (*previousResult);
}
std::cout << orderedTaskIf->getTaskName() << ": "
<< "Result = " << *((int *)(result.get())) << std::endl
<< std::flush;
return result;
}
}
};
main.cpp
#include "DerivedTasks.hpp"
// This Example simulate the execution of the following expression ((a+b)+c)*(d+e) where a,b,c,d,e are value 2
int main()
{
std::cout << "Running Example 7 for MTL Version " << MTL_VERSION_MAJOR << "." << MTL_VERSION_MINOR << "." << MTL_VERSION_PATCH << std::endl;
Var2Task var2Task; // Task that return 2
SumTask sum2Task; // Task that return the sum of the predecessors
MulTask mul2Task; // Task that return the multiplication of the predecessors
auto taskA = std::make_shared<MTL::MTLOrderedTask>("a", var2Task); // Task that return 2
auto taskB = std::make_shared<MTL::MTLOrderedTask>("b", var2Task); // Task that return 2
auto taskC = std::make_shared<MTL::MTLOrderedTask>("c", var2Task); // Task that return 2
auto taskD = std::make_shared<MTL::MTLOrderedTask>("d", var2Task); // Task that return 2
auto taskE = std::make_shared<MTL::MTLOrderedTask>("e", var2Task); // Task that return 2
auto taskF = std::make_shared<MTL::MTLOrderedTask>("f", sum2Task); // Task that return the sum of the predecessors
auto taskG = std::make_shared<MTL::MTLOrderedTask>("g", sum2Task); // Task that return the sum of the predecessors
auto taskH = std::make_shared<MTL::MTLOrderedTask>("h", sum2Task); // Task that return the sum of the predecessors
auto taskI = std::make_shared<MTL::MTLOrderedTask>("i", mul2Task); // Task that return the multiplication of the predecessors
MTL::MTLTaskFlow taskFlow; // Task Flow
taskFlow.precede(taskA, taskF); // Task A precede Task F
taskFlow.precede(taskB, taskF); // Task B precede Task F
taskFlow.precede(taskC, taskG); // Task C precede Task G
taskFlow.precede(taskF, taskG); // Task F precede Task G
taskFlow.precede(taskD, taskH); // Task D precede Task H
taskFlow.precede(taskE, taskH); // Task E precede Task H
taskFlow.precede(taskG, taskI); // Task G precede Task I
taskFlow.precede(taskH, taskI); // Task H precede Task I
taskFlow.run(); // Run the task flow
std::shared_ptr<void> result = taskFlow.getResult(); // Get the result of the task flow
std::cout << "Result: " << *(static_cast<int *>(result.get())) << std::endl; // Print the result
return 0;
}
For more examples, please refer to the Example Directory
Contributions are what make the open source community such an amazing place to learn, inspire, and create. Any contributions you make are greatly appreciated.
If you have a suggestion that would make this better, please fork the repo and create a pull request. You can also simply open an issue with the tag "enhancement". Don't forget to give the project a star! Thanks again! Before these steps please read the Contributing Guidelines and the Code of Conduct
- Fork the Project
- Create your Feature Branch (
git checkout -b feature/AmazingFeature) - Commit your Changes (
git commit -m 'Add some AmazingFeature') - Push to the Branch (
git push origin feature/AmazingFeature) - Open a Pull Request
Distributed under the GPL-3.0 License. See LICENSE for more information.
You can find more detailed documentation at MTL Documentation
ZigRazor - [email protected]
Project Link: https://github.com/ZigRazor/MTL