epics/pvxs
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

client Add Operation::wait()

Browse Source
This commit is contained in:
Michael Davidsaver
2020-03-12 19:52:00 -07:00
parent 92228556d7
commit 25e712eb2a
6 changed files with 184 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client.cpp
+56
View File
@@ -11,6 +11,7 @@
#include <osiSock.h>
#include <dbDefs.h>
#include <epicsThread.h>
#include <epicsGuard.h>
#include <pvxs/log.h>
#include <clientimpl.h>
@@ -19,6 +20,9 @@ DEFINE_LOGGER(setup, "pvxs.client.setup");
DEFINE_LOGGER(io, "pvxs.client.io");
DEFINE_LOGGER(duppv, "pvxs.client.dup");
typedef epicsGuard<epicsMutex> Guard;
typedef epicsGuardRelease<epicsMutex> UnGuard;
namespace pvxs {
namespace client {
@@ -52,6 +56,16 @@ Connected::Connected(const std::string& peerName)
Connected::~Connected() {}
Interrupted::Interrupted()
:std::runtime_error ("Interrupted")
{}
Interrupted::~Interrupted() {}
Timeout::Timeout()
:std::runtime_error ("Interrupted")
{}
Timeout::~Timeout() {}
Channel::Channel(const std::shared_ptr<Context::Pvt>& context, const std::string& name, uint32_t cid)
:context(context)
,name(name)
@@ -117,6 +131,35 @@ void Channel::disconnect(const std::shared_ptr<Channel>& self)
}
Value ResultWaiter::wait(double timeout)
{
Guard G(lock);
while(outcome==Busy) {
UnGuard U(G);
if(!notify.wait(timeout))
throw Timeout();
}
if(outcome==Done)
return result();
else
throw Interrupted();
}
void ResultWaiter::complete(Result&& result, bool interrupt)
{
bool wakeup;
{
Guard G(lock);
wakeup = outcome==Busy;
if(wakeup) {
this->result = std::move(result);
outcome = interrupt ? Abort : Done;
}
}
if(wakeup)
notify.trigger();
}
OperationBase::OperationBase(operation_t op, const std::shared_ptr<Channel>& chan)
:Operation(op)
,chan(chan)
@@ -124,6 +167,19 @@ OperationBase::OperationBase(operation_t op, const std::shared_ptr<Channel>& cha
OperationBase::~OperationBase() {}
Value OperationBase::wait(double timeout)
{
if(!waiter)
throw std::logic_error("Operation has custom .result() callback");
return waiter->wait(timeout);
}
void OperationBase::interrupt()
{
if(waiter)
waiter->complete(Result(), true);
}
RequestInfo::RequestInfo(uint32_t sid, uint32_t ioid, std::shared_ptr<OperationBase>& handle)
:sid(sid)
,ioid(ioid)
src/clientget.cpp
+16 -3
View File
@@ -116,6 +116,18 @@ struct GPROp : public OperationBase
_cancel(true);
}
void setDone(decltype (done)&& cb)
{
if(cb) {
done = std::move(cb);
} else {
auto waiter = this->waiter = std::make_shared<ResultWaiter>();
done = [waiter](Result&& result) {
waiter->complete(std::move(result), false);
};
}
}
void notify() {
try {
if(done)
@@ -429,7 +441,7 @@ std::shared_ptr<Operation> GetBuilder::_exec_get()
auto chan = Channel::build(ctx, _name);
auto op = std::make_shared<GPROp>(Operation::Get, chan);
op->done = std::move(_result);
op->setDone(std::move(_result));
op->pvRequest = _build();
chan->pending.push_back(op);
@@ -452,7 +464,8 @@ std::shared_ptr<Operation> PutBuilder::exec()
auto chan = Channel::build(ctx, _name);
auto op = std::make_shared<GPROp>(Operation::Put, chan);
op->done = std::move(_result);
op->setDone(std::move(_result));
if(_builder) {
op->builder = std::move(_builder);
} else if(pvt) {
@@ -483,7 +496,7 @@ std::shared_ptr<Operation> RPCBuilder::exec()
auto chan = Channel::build(ctx, _name);
auto op = std::make_shared<GPROp>(Operation::RPC, chan);
op->done = std::move(_result);
op->setDone(std::move(_result));
op->rpcarg = std::move(_argument);
op->pvRequest = _build();
src/clientimpl.h
+22
View File
@@ -9,6 +9,8 @@
#include <list>
#include <epicsTime.h>
#include <epicsEvent.h>
#include <epicsMutex.h>
#include <pvxs/client.h>
@@ -23,17 +25,37 @@ namespace client {
struct Channel;
struct ResultWaiter {
epicsMutex lock;
epicsEvent notify;
Result result;
enum {
Busy,
Done,
Abort,
} outcome = Busy;
Value wait(double timeout=-1.0);
void complete(Result&& result, bool interrupt);
};
// internal actions on an Operation
struct OperationBase : public Operation
{
std::shared_ptr<Channel> chan;
uint32_t ioid;
Value result;
bool done;
std::shared_ptr<ResultWaiter> waiter;
OperationBase(operation_t op, const std::shared_ptr<Channel>& chan);
virtual ~OperationBase();
virtual void createOp() =0;
virtual void disconnected(const std::shared_ptr<OperationBase>& self) =0;
virtual Value wait(double timeout=-1.0) override final;
virtual void interrupt() override final;
};
struct RequestInfo {
src/clientintrospect.cpp
+9 -1
View File
@@ -174,7 +174,15 @@ std::shared_ptr<Operation> GetBuilder::_exec_info()
auto chan = Channel::build(ctx, _name);
auto op = std::make_shared<InfoOp>(chan);
op->done = std::move(_result);
if(_result) {
op->done = std::move(_result);
} else {
auto waiter = op->waiter = std::make_shared<ResultWaiter>();
op->done = [waiter](Result&& result) {
waiter->complete(std::move(result), false);
};
}
chan->pending.push_back(op);
chan->createOperations();
src/pvxs/client.h
+63 -12
View File
@@ -60,6 +60,18 @@ struct PVXS_API Connected : public std::runtime_error
const epicsTime time;
};
struct PVXS_API Interrupted : public std::runtime_error
{
Interrupted();
virtual ~Interrupted();
};
struct PVXS_API Timeout : public std::runtime_error
{
Timeout();
virtual ~Timeout();
};
//! Holder for a Value or an exception
class Result {
Value _result;
@@ -101,6 +113,26 @@ struct PVXS_API Operation {
//! Explicitly cancel a pending operation.
//! Blocks until an in-progress callback has completed.
virtual void cancel() =0;
/** @brief Block until Operation completion
*
* As an alternative to a .result() callback, wait for operation competion,
* timeout, or interruption (via. interrupt() ).
*
* @param timeout Time to wait prior to throwing TimeoutError. cf. epicsEvent::wait(double)
* @return result Value. Always empty/invalid for put()
* @throws Timeout Timeout exceeded
* @throws Interrupted interrupt() called
*/
virtual Value wait(double timeout) =0;
//! wait(double) without a timeout
Value wait() {
return wait(99999999.0);
}
//! Queue an interruption of a wait() or wait(double) call.
virtual void interrupt() =0;
};
//! Handle for monitor subscription
@@ -175,6 +207,13 @@ public:
const Config& config() const;
/** Request the present value of a PV
*
* @code
* Context ctxt(...);
* auto result = ctxt.get("pv:name")
* .exec()
* .wait();
* @endcode
*
* @code
* Context ctxt(...);
@@ -183,6 +222,7 @@ public:
* std::cout<<prototype();
* })
* .exec();
* // store op until completion
* @endcode
*/
inline
@@ -193,11 +233,19 @@ public:
*
* @code
* Context ctxt(...);
* auto result = ctxt.info("pv:name")
* .exec()
* .wait();
* @endcode
*
* @code
* Context ctxt(...);
* auto op = ctxt.info("pv:name")
* .result([](Result&& prototype){
* std::cout<<prototype();
* })
* .exec();
* // store op until completion
* @endcode
*/
inline
@@ -209,18 +257,10 @@ public:
*
* @code
* Context ctxt(...);
* auto op = ctxt.put("pv:name")
* .set("value", 42)
* .result([](Result&& prototype){
* try {
* // always returns empty Value on success
* prototype();
* std::cout<<"Success";
* }catch(std::exception& e){
* std::cout<<"Error: "<<e.what();
* }
* })
* .exec();
* auto result = ctxt.put("pv:name")
* .set("value", 42)
* .exec()
* .wait();
* @endcode
*
* Alternately, and more generally, using a .build() callback
@@ -243,12 +283,21 @@ public:
* }
* })
* .exec();
* // store op until completion
* @endcode
*/
inline
PutBuilder put(const std::string& pvname);
/** Execute "stateless" remote procedure call operation.
*
* @code
* Value arg = ...;
* Context ctxt(...);
* auto result = ctxt.rpc("pv:name", arg)
* .exec()
* .wait();
* @endcode
*
* @code
* Value arg = ...;
@@ -258,6 +307,7 @@ public:
* std::cout<<prototype();
* })
* .exec();
* // store op until completion
* @endcode
*/
inline
@@ -277,6 +327,7 @@ public:
* }
* })
* .exec();
* // store op until completion
* @endcode
*/
inline
test/testget.cpp
+18 -1
View File
@@ -43,6 +43,22 @@ struct Tester {
initial["value"] = 42;
}
void testWaiter()
{
testShow()<<__func__;
mbox.open(initial);
serv.start();
auto op = cli.get("mailbox").exec();
cli.hurryUp();
auto result = op->wait(5.0);
testEq(result["value"].as<int32_t>(), 42);
}
void testWait()
{
client::Result actual;
@@ -217,8 +233,9 @@ void testError(bool phase)
MAIN(testget)
{
testPlan(13);
testPlan(14);
logger_config_env();
Tester().testWaiter();
Tester().loopback();
Tester().lazy();
Tester().timeout();