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

fixup client operation object lifetime

Browse Source
This commit is contained in:
Michael Davidsaver
2020-04-12 23:29:30 -07:00
parent bcea4f032a
commit b0eecb949f
5 changed files with 127 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/clientget.cpp
+46 -22
View File
@@ -131,6 +131,7 @@ struct GPROp : public OperationBase
:OperationBase (op, chan)
{}
~GPROp() {
chan->context->tcp_loop.assertInLoop();
_cancel(true);
}
@@ -161,32 +162,33 @@ struct GPROp : public OperationBase
}
}
virtual void cancel() override final {
_cancel(false);
}
void _cancel(bool implicit) {
virtual void cancel() override final
{
auto context = chan->context;
decltype (done) junk;
context->tcp_loop.call([this, &junk, implicit](){
if(implicit && state!=Done) {
log_warn_printf(setup, "implied cancel of op%x on channel '%s'\n",
op, chan ? chan->name.c_str() : "");
}
if(state==GetOPut || state==Exec) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
}
state = Done;
chan.reset();
context->tcp_loop.call([this, &junk](){
_cancel(false);
junk = std::move(done);
// leave opByIOID for GC
});
}
void _cancel(bool implicit) {
if(implicit && state!=Done) {
log_warn_printf(setup, "implied cancel of op%x on channel '%s'\n",
op, chan ? chan->name.c_str() : "");
}
if(state==GetOPut || state==Exec) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
}
state = Done;
}
virtual void createOp() override final
{
if(state!=Connecting) {
@@ -450,6 +452,27 @@ void Connection::handle_GET() { handle_GPR(CMD_GET); }
void Connection::handle_PUT() { handle_GPR(CMD_PUT); }
void Connection::handle_RPC() { handle_GPR(CMD_RPC); }
static
void gpr_cleanup(std::shared_ptr<Operation>& ret, std::shared_ptr<GPROp>&& op)
{
auto cap(std::move(op));
ret.reset(cap.get(), [cap](Operation*) mutable {
// from use thread
cap->chan->context->tcp_loop.call([&cap]() {
auto temp(std::move(cap));
// on worker
try {
temp->_cancel(true);
}catch(std::exception& e){
log_exc_printf(setup, "Channel %s error in get cancel(): %s",
temp->chan->name.c_str(), e.what());
}
// ensure dtor on worker
temp.reset();
});
});
}
std::shared_ptr<Operation> GetBuilder::_exec_get()
{
std::shared_ptr<Operation> ret;
@@ -465,7 +488,8 @@ std::shared_ptr<Operation> GetBuilder::_exec_get()
chan->pending.push_back(op);
chan->createOperations();
ret = std::move(op);
gpr_cleanup(ret, std::move(op));
assert(ret);
});
return ret;
@@ -503,7 +527,7 @@ std::shared_ptr<Operation> PutBuilder::exec()
chan->pending.push_back(op);
chan->createOperations();
ret = std::move(op);
gpr_cleanup(ret, std::move(op));
});
return ret;
@@ -532,7 +556,7 @@ std::shared_ptr<Operation> RPCBuilder::exec()
chan->pending.push_back(op);
chan->createOperations();
ret = std::move(op);
gpr_cleanup(ret, std::move(op));
});
return ret;
src/clientimpl.h
+1 -1
View File
@@ -42,7 +42,7 @@ struct ResultWaiter {
// internal actions on an Operation
struct OperationBase : public Operation
{
std::shared_ptr<Channel> chan;
const std::shared_ptr<Channel> chan;
uint32_t ioid;
Value result;
bool done;
src/clientintrospect.cpp
+33 -19
View File
@@ -36,35 +36,35 @@ struct InfoOp : public OperationBase
virtual ~InfoOp()
{
chan->context->tcp_loop.assertInLoop();
_cancel(true);
}
virtual void cancel() override final {
_cancel(false);
}
void _cancel(bool implicit) {
auto context = chan->context;
decltype (done) junk;
context->tcp_loop.call([this, &junk, implicit](){
if(implicit && state!=Done) {
log_warn_printf(setup, "implied cancel of INFO on channel '%s'\n",
chan ? chan->name.c_str() : "");
}
if(state==Waiting) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
}
state = Done;
chan.reset();
context->tcp_loop.call([this, &junk](){
_cancel(false);
junk = std::move(done);
// leave opByIOID for GC
});
}
void _cancel(bool implicit) {
if(implicit && state!=Done) {
log_warn_printf(setup, "implied cancel of INFO on channel '%s'\n",
chan ? chan->name.c_str() : "");
}
if(state==Waiting) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
}
state = Done;
}
virtual void createOp() override final
{
if(state!=Connecting) {
@@ -189,7 +189,21 @@ std::shared_ptr<Operation> GetBuilder::_exec_info()
chan->pending.push_back(op);
chan->createOperations();
ret = std::move(op);
ret.reset(op.get(), [op](Operation*) mutable {
// on user thread
auto temp(std::move(op));
temp->chan->context->tcp_loop.call([&temp]() {
// on worker
try {
temp->_cancel(true);
}catch(std::exception& e){
log_exc_printf(setup, "Channel %s error in info cancel(): %s",
temp->chan->name.c_str(), e.what());
}
// ensure dtor on worker
temp.reset();
});
});
});
return ret;
src/clientmon.cpp
+41 -29
View File
@@ -72,6 +72,7 @@ struct SubscriptionImpl : public OperationBase, public Subscription
,ackTick(event_new(chan->context->tcp_loop.base, -1, EV_TIMEOUT, &tickAckS, this))
{}
virtual ~SubscriptionImpl() {
chan->context->tcp_loop.assertInLoop();
_cancel(true);
}
@@ -92,8 +93,6 @@ struct SubscriptionImpl : public OperationBase, public Subscription
virtual void pause(bool p) override final
{
if(!chan)
return;
chan->context->tcp_loop.call([this, p](){
log_info_printf(io, "Server %s channel %s monitor %s\n",
chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
@@ -165,39 +164,38 @@ struct SubscriptionImpl : public OperationBase, public Subscription
}
virtual void cancel() override final {
_cancel(false);
}
void _cancel(bool implicit) {
auto context = chan->context;
decltype (event) junk;
context->tcp_loop.call([this, &junk, implicit](){
if(implicit && state!=Done) {
log_info_printf(io, "Server %s channel %s monitor implied cancel\n",
chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
chan->name.c_str());
}
log_info_printf(io, "Server %s channel %s monitor cancel\n",
chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
chan->name.c_str());
if(state==Idle || state==Running) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
if(pipeline)
(void)event_del(ackTick.get());
}
state = Done;
chan.reset();
context->tcp_loop.call([this, &junk](){
_cancel(false);
junk = std::move(event);
// leave opByIOID for GC
});
}
void _cancel(bool implicit) {
if(implicit && state!=Done) {
log_info_printf(io, "Server %s channel %s monitor implied cancel\n",
chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
chan->name.c_str());
}
log_info_printf(io, "Server %s channel %s monitor cancel\n",
chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
chan->name.c_str());
if(state==Idle || state==Running) {
chan->conn->sendDestroyRequest(chan->sid, ioid);
// This opens up a race with an in-flight reply.
chan->conn->opByIOID.erase(ioid);
chan->opByIOID.erase(ioid);
if(pipeline)
(void)event_del(ackTick.get());
}
state = Done;
}
virtual void createOp() override final
{
if(state!=Connecting) {
@@ -559,7 +557,21 @@ std::shared_ptr<Subscription> MonitorBuilder::exec()
chan->pending.push_back(op);
chan->createOperations();
ret = std::move(op);
ret.reset(op.get(), [op](Subscription*) mutable {
// on user thread
auto temp(std::move(op));
temp->chan->context->tcp_loop.call([&temp]() {
// on worker
try {
temp->_cancel(true);
}catch(std::exception& e){
log_exc_printf(monevt, "Channel %s error in monitor cancel(): %s",
temp->channelName.c_str(), e.what());
}
// ensure dtor on worker
temp.reset();
});
});
});
return ret;
src/evhelper.cpp
+6 -1
View File
@@ -249,7 +249,12 @@ void evbase::call(std::function<void()>&& fn)
void evbase::assertInLoop()
{
assert(pvt->worker.isCurrentThread());
if(!pvt->worker.isCurrentThread()) {
char name[32];
pvt->worker.getName(name, sizeof(name));
log_exc_printf(logerr, "Not in evbase working: \"%s\" != \"%s\"\n",
name, epicsThread::getNameSelf());
}
}
bool evbase::inLoop()