benlau / asyncfuture Goto Github PK
View Code? Open in Web Editor NEWUse QFuture like a Promise object
License: Apache License 2.0
Use QFuture like a Promise object
License: Apache License 2.0
This is a very handy library - thank you. But:
Consider this function:
QFuture<QList<double>> foo() {
AsyncFuture::Deferred<QList<double>> _terrain_fetched;
_terrain_fetched.complete(QList<double>());
return _terrain_fetched.future();
}
and the compiler error:
In file included from ../foo.h:7,
from ../foo.cpp:4:
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h: In instantiation of ‘void AsyncFuture::Private::DeferredFuture<T>::reportResult(QList<R>&) [with R = double; T = QList<double>]’:
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:505:9: required from ‘void AsyncFuture::Private::DeferredFuture<T>::complete(QList<R>&) [with R = double; T = QList<double>]’
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:1280:9: required from ‘void AsyncFuture::Deferred<T>::complete(T) [with T = QList<double>]’
../foo.cpp:78:46: required from here
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:661:46: error: no matching function for call to ‘AsyncFuture::Private::DeferredFuture<QList<double> >::reportResult(double&, int&)’
QFutureInterface<T>::reportResult(value[i], i);
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^~~~~~~~~~~~~
In file included from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfuture.h:45,
from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/QtCore:85,
from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtConcurrent/QtConcurrentDepends:3,
from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtConcurrent/QtConcurrent:3,
from ../foo.cpp:1:
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:189:13: note: candidate: ‘void QFutureInterface<T>::reportResult(const T*, int) [with T = QList<double>]’
inline void QFutureInterface<T>::reportResult(const T *result, int index)
^~~~~~~~~~~~~~~~~~~
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:189:13: note: no known conversion for argument 1 from ‘double’ to ‘const QList<double>*’
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:209:13: note: candidate: ‘void QFutureInterface<T>::reportResult(const T&, int) [with T = QList<double>]’
inline void QFutureInterface<T>::reportResult(const T &result, int index)
^~~~~~~~~~~~~~~~~~~
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:209:13: note: no known conversion for argument 1 from ‘double’ to ‘const QList<double>&’
../foo.cpp:43:40: warning: ‘airmap::geography::bounding_box fromQt(const QGeoRectangle&)’ defined but not used [-Wunused-function]
static airmap::geography::bounding_box fromQt(const QGeoRectangle& bbox) {
for some reason the compiler prefers to match this instead of this.
It would be nice to support types with deleted copy constructor/assign operator,
for AsyncFuture::Observable
and AsyncFuture::Deferred
instantation, for example:
struct Foo {
explicit Foo(std::string s): s_{s} {}
Foo(const Foo &) = delete;
Foo &operator=(const Foo &) = delete;
Foo(Foo &&o): s_{std::move(o.s_)} { }
Foo &operator=(Foo &&o) {
s_ = std::move(o.s_);
return *this;
}
...
};
If a chain of futures is cancelled with the first future finished by a signal the inner subscribe onCancelled callback on a future of signal is not called, but the outer future is cancelled correctly.
Expected:
The onCancelled of the current running inner future is called
Note:
This test is based on fork of @vpicaver asyncfuture on commit @0da13489746f0412e6bad66bfe0acef54915adb2
class TestClass : public QObject
{
Q_OBJECT
public:
signals:
void someTestSignal();
void someOtherTestSignal();
};
TEST_F(AsyncFutureTest, futureFromTwoSignalsWithFirstSignalCalledShouldBeCancelled)
{
TestClass testclass;
auto deferWasCancelled = AsyncFuture::deferred<bool>();
auto innerDeferWasCancelled = AsyncFuture::deferred<bool>();
auto future = AsyncFuture::observe(&testclass, &TestClass::someTestSignal)
.subscribe(
[&innerDeferWasCancelled, &testclass]() {
return AsyncFuture::observe(&testclass,
&TestClass::someOtherTestSignal)
.subscribe(
[&innerDeferWasCancelled]() {
innerDeferWasCancelled.complete(false);
},
[&innerDeferWasCancelled]() {
//following deferred gets never called
innerDeferWasCancelled.complete(true);
});
},
[&deferWasCancelled]() { deferWasCancelled.complete(true); })
.future();
emit testclass.someTestSignal();
QTimer::singleShot(40, [&future]() { future.cancel(); });
waitUntil(future);
ASSERT_TRUE(future.isCanceled());
ASSERT_TRUE(future.isFinished());
//following waitUntil gets never finished
waitUntil(innerDeferWasCancelled.future());
ASSERT_TRUE(innerDeferWasCancelled.future().result());
waitUntil(deferWasCancelled.future());
ASSERT_TRUE(deferWasCancelled.future().result());
}
As you probably know, this library doesn't work with Qt6. I'm currently working to upgrade one of my projects from Qt5 to Qt6, so I'm wondering if plans exist to upgrade this library to support it? Or if I should port my code to use something else?
Thanks in advance for taking a peek!
An example of this issue is if two futures being added to Combined(). If future1 finished before future2 has been added, this will complete() the Combined(). Even though in this use case, we want the code below to print out "Finished!" once both future1 and future2 are finished. If future1 is finished before future2 is added, Finished() will be prematurely, called.
auto combine = Combined() << future1 << future2;
combine.subscribe([]() { qDebug() << "Finished!"; }
I don't think this can be easily be fixed without api changes to how Combine() is handled. Perhaps it better to advocate using async on the main thread only. The deferred needs to have a proper context object to make this work correrctly. By default deferred is move to the main thread.
Now that Qt 6.3 supports whenAll (https://doc-snapshots.qt.io/qt6-6.3/qtfuture.html#whenAll) we should be able to port asyncfuture's backend to use Qt 6.3 API. That's probably performance benefits since Qt QFuture api doesn't use QObject to combine futures.
I was a bit surprised to find that this doesn't work:
Observable<void> o;
auto all = combine() << o;
sure you can just call o.future()
but it seems like that should be overloaded in combine.
Could this just be added to the Combinator
?
template <typename T>
Combinator& operator<<(Observable<T> observable)
{
combinedFuture->addFuture(observable.future());
return *this;
}
It would be nice if the README.md has documentation on how to build the unit testcases. They currently require downloading qpm and installing the dependancies which aren't documented.
what dependencies that I have to install ?
And how does one communicate that error?
Is this what 'cancel' is for? - doesn't sound this way as cancel is the interface of the future not of the promise.
And a related question, having a failed QFuture
- how does one "rescue" it? - i.e.: having this chain:
return observe(reading).subscribe(updateCache).future();
- suppose reading
fails and suppose there is a way of correcting it:
return observe(reading).subscribe(updateCache, rescue).future();
the above doesn't work.
Chaining asyncfuture is useful to chain QFuture's together. Sometimes at the begin of the chain, it's useful to have completed QFuture that come from a fixed value. To achieve this with asyncfuture, a programmer needs to use deferred() and then complete it with deferred::complete(). Although this works, deferred creates a Deferred object and a that holds a DeferredFuture. A DeferredFuture is a fairly heavy weight object since inherits QObject and hold a mutex. A DeferrerFuture is usually stored as a shared object and require reference counting.
To reduce this overhead for the specific job having a QFuture that's automatically complete, I propose having a QFuture<void> finished()
and QFuture<T> finished<T>() functions.
They would be implement as the following:
inline QFuture<void> finished() {
QFutureInterface<void> fi;
fi.reportFinished();
return QFuture<void>(&fi);
}
template <typename T> QFuture<T> finished(const T &val) {
QFutureInterface<T> fi;
fi.reportFinished(&val);
return QFuture<T>(&fi);
}
With Qt 5.14.1, AsyncFuture testcases produce many warnings about using deprecated qt functions. These should be update so AsyncFuture continues to work in the future.
For example this pipeline should work:
Run 100 concurrent jobs, then setup another 100 jobs, finally setup another 100 jobs.
The user should be able to get the correct progress by watching the last future in this pipeline. Below is a test-case.
QVector<int> ints(100);
std::iota(ints.begin(), ints.end(), ints.size());
std::function<int (int)> func = [](int x)->int {
QThread::msleep(100);
return x * x;
};
QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);
auto nextFuture = AsyncFuture::observe(mappedFuture).subscribe([ints, func](){
QFuture<int> mappedFuture2 = QtConcurrent::mapped(ints, func);
return mappedFuture2;
}).future();
bool nextExecuted2 = false;
auto nextFuture2 = AsyncFuture::observe(nextFuture).subscribe([&nextExecuted2, ints, func](){
QFuture<int> mappedFuture2 = QtConcurrent::mapped(ints, func);
nextExecuted2 = true;
return mappedFuture2;
}).future();
int progress = -1;
AsyncFuture::observe(nextFuture2).onProgress([&progress, nextFuture2](){
QVERIFY2(progress <= nextFuture2.progressValue(), QString("%1 <= %2").arg(progress).arg(nextFuture2.progressValue()).toLocal8Bit());
progress = nextFuture2.progressValue();
});
await(nextFuture2);
QCOMPARE(nextFuture2.progressMinimum(), 0);
QCOMPARE(nextFuture2.progressMaximum(), ints.size() * 3);
QCOMPARE(nextExecuted2, true);
QCOMPARE(nextFuture2.progressValue(), ints.size() * 3);
When setting up a chain of futures a common use case is to have a single cancellation token associated with the chain such that if it is ever canceled then the chain will short circuit and not execute any more of the chain by unsubscribing the next in sequence.
Here's a small example:
auto testToken = deferred<void>();
auto firstHandler = [] { std::cout << "firstHandler called\n"; };
auto secondHandler = [testToken]() mutable {
std::cout << "secondHandler called\n";
testToken.cancel();
};
auto thirdHandler = [] { std::cout << "Oops, did not want thirdHandler to be called\n"; };
auto starter = deferred<void>();
QFuture<void> f1 = starter.future();
QFuture<void> f2 = observe(f1).subscribe(firstHandler).future();
QFuture<void> f3 = observe(f2).subscribe(secondHandler).future();
QFuture<void> f4 = observe(f3).subscribe(thirdHandler).future();
testToken.complete(f4);
testToken.subscribe([] { std::cout << "chain completed\n"; },
[] { std::cout << "chain canceled\n"; });
starter.complete();
Output:
firstHandler called
secondHandler called
chain canceled
Oops, did not want thirdHandler to be called
The code is behaving exactly as it should since f3 completes which causes f4 with the thirdHandler to be called - only the testToken was canceled, and f4 was not.
What I would like to do in a clean way is cancel the next future in the chain from within the handler, and there are many ways to do that but none of the methods I have come up with are very ergonomic.
One solution is to just pass a value indicating whether to continue or not, for example:
auto testToken = deferred<bool>();
auto firstHandler = [](bool cont) {
if (!cont)
return false;
std::cout << "firstHandler called\n";
return true;
};
auto secondHandler = [](bool cont) mutable {
if (!cont)
return false;
std::cout << "secondHandler called\n";
// cancel chain by returning false here
return false;
};
auto thirdHandler = [](bool cont) {
if (!cont)
return false;
std::cout << "Oops, did not want thirdHandler to be called\n";
return true;
};
auto starter = deferred<bool>();
QFuture<bool> f1 = starter.future();
QFuture<bool> f2 = observe(f1).subscribe(firstHandler).future();
QFuture<bool> f3 = observe(f2).subscribe(secondHandler).future();
QFuture<bool> f4 = observe(f3).subscribe(thirdHandler).future();
testToken.complete(f4);
testToken.subscribe(
[](bool cont) {
if (cont) {
std::cout << "chain completed\n";
} else {
std::cout << "chain canceled by passing value\n";
}
},
[] { std::cout << "chain canceled\n"; });
starter.complete(true);
Output:
firstHandler called
secondHandler called
chain canceled by passing value
But this solution leads to every handler always being called, and also pollutes the arguments with an unrelated bool.
There are other solutions such as creating a deferred for each place the chain may be canceled and then chaining the calls through those deferred's, or establishing the chain from within the handlers themselves (e.g. firstHandler invokes the call to secondHandler). But these solutions are very messy to work with and make the handlers less composable as functions.
Maybe I have overlooked something in the API and there is a simple way to achieve what I am looking for, any help would be appreciated.
In C# using Tasks it is done by passing a cancellation token along with the handler function when subscribing, e.g. it would look something like this:
CancellationToken token;
QFuture<void> f1 = GetSomeFuture();
QFuture<void> f2 = observe(f1).subscribe(firstHandler, token).future();
Then if anything calls token.cancel() it would actually unsubscribe any functions that were subscribed with the token. Also long running functions in the chain can decide to regularly check the token to see if token.isCanceled() to stop mid function call.
Considering this situation: Display a dialog, and then use AsyncFuture
to send a network request, and the result will be displayed on the interface when the network request responds. I used the following logic (pseudo code):
Dialog::request()
{
auto future = ... ; // request data from network
observe(future).subscribe([=](QString data){ // onComplete
ui->label->setText(data);
},
[=](){ // onCanceled
ui->errorLabel->setText("error");
});
}
But before the network reply, the user may close the dialog. In this case, the ui->label
in onComplete has been destroyed, causing a crash.
Is there any plan to add contextObject
to function subscribe
? So I can modify code like this
Dialog::request()
{
auto future = ... ; // request data from network
- observe(future).subscribe([=](QString data){ // onComplete
+ observe(future).subscribe(this, [=](QString data){ // onComplete
ui->label->setText(data);
},
[=](){ // onCanceled
ui->errorLabel->setText("error");
});
}
The following testcase should pass:
QVector<int> ints(100);
std::iota(ints.begin(), ints.end(), ints.size());
std::function<int (int)> func = [](int x)->int {
QThread::msleep(100);
return x * x;
};
QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);
QFuture<int> runFuture = QtConcurrent::run([]() {
QThread::msleep(100);
return 10;
});
AsyncFuture::Combinator combine;
combine << mappedFuture << runFuture;
auto combineFuture = combine.future();
QCOMPARE(combineFuture.progressMinimum(), 0);
QCOMPARE(combineFuture.progressMaximum(), ints.size() + 1);
int progress = 0;
AsyncFuture::observe(combineFuture).onProgress([&progress, combineFuture](){
QVERIFY(progress <= combineFuture.progressValue());
progress = combineFuture.progressValue();
});
await(combineFuture);
QCOMPARE(combineFuture.progressMaximum(), combineFuture.progressValue());
It's again about this type of usage pattern:
QFuture<double> compute(QVector<QPointF> coords) const
{
QFuture<System> sys = getSystem();
auto f = [=](System&& system)
{
return QtConcurrent::mapped(coords, Worker(system));
};
auto rv = AsyncFuture::observe(sys).subscribe(f);
return rv.future();
}
In this case, if client code cancels the future returned by compute()
, I need it to actually cancel the QtConcurrent::mapped
operation. I may be wrong but I think that this behaviour is generally desirable, as there is nothing else I can see to which the cancel()
could relate in this type of situation. The idea is that the QFuture
returned by a Observable::future()
basically "becomes" the QFuture
returned from the lambda function. However, right now cancelling the returned QFuture
has no effect at all.
To achieve it, I wrote the following:
template<typename RetType, typename InpType, typename Func>
class FutureFunctor
{
public:
explicit FutureFunctor(Func wrapped) :
_wrapped(wrapped),
_s(std::make_shared<State>())
{}
FutureFunctor(const FutureFunctor&) = default;
FutureFunctor(FutureFunctor&&) = default;
void cancel() const
{
_s->isCancelled = true;
_s->future.cancel();
}
QFuture<RetType> operator()(InpType&& input) const
{
Q_ASSERT(!_s->isCalled);
_s->isCalled = true;
if(!_s->isCancelled) {
_s->future = _wrapped(std::forward<InpType>(input));
}
return _s->future;
}
private:
struct State {
QFuture<RetType> future;
bool isCalled = false;
bool isCancelled = false;
};
const Func _wrapped;
std::shared_ptr<State> _s;
};
template<typename RetType, typename InpType, typename Func>
QFuture<RetType> apply(Func f, QFuture<InpType> input)
{
auto ff = FutureFunctor<RetType, InpType, Func>(f);
auto rv = AsyncFuture::observe(input).subscribe(ff);
rv.subscribe([]() {}, [ff]() { ff.cancel(); });
return rv.future();
}
Applying it to the use case above, it looks like so:
QFuture<double> compute(QVector<QPointF> coords) const
{
QFuture<System> sys = getSystem();
auto f = [=](System&& system)
{
return QtConcurrent::mapped(coords, Worker(system));
};
return apply<double>(f, sys);
}
As I think that this feature can be generally useful, I thought you might want to include it in the lib, either as an additional layer like in my code, or maybe also by modifying the behaviour of `subscribe in some way.
Currently asyncfuture doesn't support watching progress on work thread because the future watcher is always move to the main thread. If the work thread has an eventloop, this should work. I propose being able to send a context object into Observable::onProgress(QObject* context) to support multi threaded progress watching.
As newcomer I would like to see that testing with -fsanitize=thread
have been done.
In perfect world of course this should be part of CI, but not sure how much efforts this requires,
because of you need compile Qt from source code with -fsanitize=thread
flag.
So may be just mention in readme that such kind of testing was done for OS Z/compiler Y/Qt X and all tests are green?
If the contextObject is in a different thread than the current execution thread, asyncfuture can cause a crash. This is because QFutureWatcher lives in the current execution thread() but communicating to the contextObject's thread(). In this case, it's possible for the QFutureWatcher to emit bath canceled() and finished() and since both the signals in a queued eventloop, they are handled. By the time they're handled, QFutureWatcher has been destroyed, causing a crash in the other single handler. This isn't a problem if QFutureWatcher and contextObject live in the same thread because QFutureWatcher gets disconnected, when it's destroyed.
I was trying to load data one by one, like supported in QFuture::resultReadyAt.
Here is an example:
struct ComplexData{
QString fruit, color;
};
using ComplexDataPtr = QSharedPointer<ComplexData>;
// Simulate network reply content (QNetworkReply)
QFuture<QString> dataLoaderFuture(){
return QtConcurrent::run([](){
QList<QString> list{"Apple,red","Banana,yellow","Orange,orange"};
auto replyContent = list.join(";");
qDebug()<<"Sending reply:"<<replyContent;
return replyContent;
});
}
// The future holds multiple results, one for each fruit.
QFuture<ComplexDataPtr> getFruitsOneByOne_AsyncFuture(){
qDebug()<<"Getting fruits one by one (using AsyncFuture).";
auto defer = AsyncFuture::deferred<ComplexDataPtr>();
auto reply = dataLoaderFuture(); // returns QString "Apple,red;Banana,yellow;Orange,orange"
auto observe = AsyncFuture::observe(reply).subscribe([=]() {
qDebug()<<"Creating fruits from reply";
QList<ComplexDataPtr> results;
QString replyContent = reply.result();
QStringList records = replyContent.split(";");
auto d = defer;
int index = 0;
for ( const auto &record : records ){
results << createFruitFromRecord(record);
d.reportResult(results.last(),index++); // DOES NOT WORK AS EXPECTED
}
//d.complete(results); // THIS IS NOT WHAT I WANT IN THIS TEST CASE
});
return defer.future();
}
The Deferred
class lacks a suitable method. I naivly tried to add the method reportResult
but it did not work.
void reportStarted() {
deferredFuture->reportStarted();
}
// NOT PART OF ORIGINAL LIBRARY, ADDED BECAUSE
// I COULD NOT GET "resultReadyAt" SIGNAL TO WORK
// AND THIS WAS MY ATTEMPT TO IMPLEMENT IT (NO SUCCESS)
void reportResult(const T &result, int index){
deferredFuture->reportResult(result, index);
}
protected:
QSharedPointer<Private::DeferredFuture<T> > deferredFuture;
};
In my use case, a class maybe called to do work multiple time. It's generally useful to start an processing the data, but if new data is submitted, the class restarts the processing and cancels previous runs. For example:
class TestClass {
public:
void doWork() {
if(m_doWorkFuture.isRunning() || m_doWorkFuture.isStarted()) {
m_doWorkFuture.cancel();
}
int currentCount = m_count;
auto runFuture = QtConcurrent::run([currentCount](){
return currentCount + 1;
});
m_doWorkFuture = runFuture;
m_subscribeFuture = observe(runFuture).subscribe([this](){
m_finishCount++;
},
[this](){
m_cancelCount++;
}).future();
}
void waitToFinish() {
await(m_subscribeFuture);
}
QFuture<void> m_subscribeFuture;
QFuture<int> m_doWorkFuture;
int m_count = 0;
int m_finishCount = 0;
int m_cancelCount = 0;
};
TestClass myTest;
myTest.doWork();
myTest.doWork();
myTest.doWork();
myTest.waitToFinish();
QCOMPARE(myTest.m_finishCount, 1);
QCOMPARE(myTest.m_cancelCount, 2);`
This test currently fails on the master on windows.
Currently, this library offers a combiner which resolves once all the futures added to it have resolved.
I don't think there is a 'combiner' which resolves once ANY ONE of the futures added to it have resolved. It would be nice to have this as a feature.
Perhaps 2 versions could be implemented. One would behave like Promise.any() and the other like Promise.race().
My motivation is the following use case:
I have a result that depends on an arbitrary number of inputs, where arbitrary means: possibly also 0. Each of the inputs can either be immediately available or require delegation to a thread. Now I use AsyncFuture::Combinator
to detect when all the results are ready.
My first attempt looked like this:
[...]
auto combiner = AsyncFuture::combine();
QVector<QFuture<Result>> results;
for(auto item: items) {
auto future = getCachedResultOrCompute(item);
results << future;
combiner << future;
}
return combiner.subscribe([results]() {
// do something with the results...
}).future();
This approach has one problem though: If items
is empty, the callback given to subscribe
will never get called. So, it'd be convenient if it were possible to write something like the following wrapper to subscribe
on Combinator
:
template<typename RetType, typename Func>
inline QFuture<RetType> subscribeToCombinator(Func f, AsyncFuture::Combinator combinator)
{
auto cancelToken = AsyncFuture::deferred<void>();
if(combinator.isEmpty()) {
auto rv = AsyncFuture::deferred<RetType>();
rv.complete(RetType());
return rv.future();
}
return combinator.subscribe(f).future();
}
This requires a method isEmpty
(or whatever you'd like to call it) on Combinator, allowing the client to query whether or not there is already a QFuture
on Combinator. As far as I can see, it is easily implented by adding the following code to Combinator
:
bool isEmpty() const { return combinedFuture->count == 0; }
Hello,
Could please add Automator class in the repo
Thanks
DeferredFuture inherits QObject, but uses a sharedpointer to do reference count when copying a QFuture around. If the QFuture is copied to another thread and the reference count goes to zero. The DeferredFuture is delete in a seperate thread that is not equal to the QObject's thread. This is bad because, QObject needs to be deleted in the thread it lives in. When the QObject is signal from it's thread, but is delete from a seperate thread, it crashes the program. This happens intermittently, but using Qt with address sanitation turned on, will easily show the issue.
When DeferredFuture's need to get delete, they just need to use a QMetaObject::invokeMethod to do the deletion on their thread.
I'll upload a Pull request and a testcase soon to demonstrate the issue.
If a combine is created, the future() return from it should cancel all the child futures, if it's canceled.
The following testcase should pass:
void BugTests::test_combine_forward_cancel() {
QAtomicInt runCount = 0;
const int count = 100;
QThreadPool::globalInstance()->setMaxThreadCount(2);
auto createMappedFuture = [&runCount, count]() {
QVector<int> ints(count);
std::iota(ints.begin(), ints.end(), ints.size());
std::function<int (int)> func = [&runCount](int x)->int {
QThread::msleep(10);
runCount++;
return x * x;
};
QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);
return mappedFuture;
};
QList<QFuture<int>> futures;
for(int i = 0; i < 4; i++) {
futures.append(createMappedFuture());
}
auto c = combine();
c << futures;
auto combineFuture = c.future();
observe(combineFuture).onProgress([&combineFuture](){
//Cancel after startup
combineFuture.cancel();
});
await(combineFuture);
QThreadPool::globalInstance()->waitForDone();
//All the sub futures should be canceled
for(auto future : futures) {
QVERIFY(future.isCanceled());
}
//Make sure the sub futures didn't run all the way through
double fullNumberOfRuns = futures.size() * count;
double ratio = runCount / fullNumberOfRuns;
QVERIFY(ratio < 0.1);
QThreadPool::globalInstance()->setMaxThreadCount(QThread::idealThreadCount());
}
Hello,
I've got one specific use case that could not resolve. Here is the test:
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <asyncfuture/asyncfuture.h>
namespace {
QFuture<void> createCompletedFuture() {
return {};
}
QFuture<void> createNestedFuture() {
auto deferred = AsyncFuture::Deferred<void>();
AsyncFuture::observe(createCompletedFuture()).subscribe([=]() mutable {
deferred.complete();
});
return deferred.future();
}
}
TEST(NestedAsyncFutureTest, Testing)
{
auto future = createNestedFuture();
future.waitForFinished();
}
This test never finishes.
I cannot use complete(future)
API because inside of subscribe() lambda I also need to emit some signals in the real code.
Is there a workaround for this? or is this a bug?
Thank you!
I have a test , but got some errors.
#include "asyncfuture.h"
#include "automator.h"
SnapshotWidget snapshotwidget;
QFuture<void> m_subscribeFuture;
QFuture<void> m_doWorkFuture;
int m_count = 0;
TaskCenter::TaskCenter(){}
void TaskCenter::realtime_snapshot_update(int interval) {
if(m_doWorkFuture.isRunning() || m_doWorkFuture.isStarted()) {
m_doWorkFuture.cancel();
}
QFuture<void> runFuture;
for (int i = 0 ; i < 50000;i++) {
runFuture = QtConcurrent::run(&this->snapshotwidget, static_cast<void(SnapshotWidget::*)()>(&SnapshotWidget::dynamic_update_snapshot));
m_doWorkFuture = runFuture;
m_subscribeFuture = AsyncFuture::observe(m_doWorkFuture).subscribe([this](){
qDebug() << "m_count: " << m_count;
m_count++;
},
[this](){
//
}).future();
Automator::wait(interval * 1000);
}
return;
}
QWidget* anotherwdt = new QWidget();
void SnapshotWidget::dynamic_update_snapshot()
{
if(!ui->onewdt->layout()){
FlowLayout *anotherwdtlay = new FlowLayout();
anotherwdt->setLayout(anotherwdtlay);
QScrollArea *scrollarea = new QScrollArea();
scrollarea->setWidget(anotherwdt);
QVBoxLayout* onewdtlay = new QVBoxLayout();
ui->onewdt->setLayout(onewdtlay);
ui->onewdt->layout()->addWidget(scrollarea);
}
QLayout* anotherwdtlay = anotherwdt->layout();
qDeleteAll(anotherwdt->findChildren<QCustomPlot *>(QString()));
QCustomPlot* plot = new QCustomPlot();
...
...
anotherwdtlay->addWidget(plot);
return;
}
the error messages are in dynamic_update_snapshot:
QObject::setParent: Cannot set parent, new parent is in a different thread
and the child widgets din't change.
please help me how to modify the codes?
thanks!
In my use case, a function compute()
that is called by UI classes returns a QFuture
. The UI classes use a QFutureWatcher
to update their state. They also depend on the started()
signal, showing a progress bar when a computation is going on.
The calculation performed inside compute()
consists basically of two parts, the first of which may or may not be cached. It looks about like this:
QFuture<double> compute(QVector<QPointF> coords) const
{
QFuture<System> as = getSystem();
auto f = [=](System&& system)
{
return QtConcurrent::mapped(std::move(coords), Worker(system));
};
auto rv = AsyncFuture::observe(as).subscribe(f);
return rv.future();
}
So I needed the future being returned to trigger the started()
signal at some point. The following patch to asyncfuture.h
worked for me:
void complete(QFuture<T> future) {
+ QFutureInterface<T>::reportStarted();
+
incRefCount();
auto onFinished = [=]() {
this->completeByFinishedFuture<T>(future);
this->decRefCount();
};
...
This feels a little bit like a hack and I'm not sure when exactly the started()
signal should be emitted for the most generally useful behaviour, but as the QtConcurrent implementations of QFuture always trigger started, I think DeferredFuture should also do this at some point for consistency.
It may be a good idea to add exception handling in a way similar to QtConcurrent.
Something like the following patch may do the trick:
---------------------- 3rdparty/asyncfuture/asyncfuture.h ----------------------
index 9abf14e..a658653 100644
@@ -960,18 +960,23 @@ eval(Functor functor, QFuture<T> future) {
template <typename DeferredType, typename RetType, typename T, typename Completed, typename Canceled>
static DeferredFuture<DeferredType>* execute(QFuture<T> future, QObject* contextObject, Completed onCompleted, Canceled onCanceled) {
DeferredFuture<DeferredType>* defer = new DeferredFuture<DeferredType>();
defer->autoDelete = true;
watch(future,
contextObject,
contextObject,[=]() {
- Value<RetType> value = eval(onCompleted, future);
- defer->complete(value);
+ try {
+ Value<RetType> value = eval(onCompleted, future);
+ defer->complete(value);
+ } catch(const QException& e) {
+ defer->reportException(e);
+ } catch(const std::exception&) {
+ defer->reportException(QUnhandledException());
}, [=]() {
onCanceled();
defer->cancel();
});
if (contextObject) {
defer->cancel(contextObject, &QObject::destroyed);
}
As far as I can see Observable<..>::context
could and should take a const QObject*
, as QObject::connect(..)
does as well.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.