Replace exit(1) calls in client with on_fatal_error_hook

src/client/cluster_client.cpp

@@ -14,7 +14,13 @@
 
 cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config, std::unique_ptr<etcd_state_client_t> st_cli_ptr)
 {
+    auto on_fatal_error_hook = [](int, const std::string& error)
+    {
+        throw std::runtime_error(error);
+    };
+
     wb = new writeback_cache_t();
+    wb->parent = this;
 
     cli_config = config.object_items();
     file_config = osd_messenger_t::read_config(config);
@@ -23,12 +29,13 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
     this->ringloop = ringloop;
     this->tfd = tfd;
 
-    msgr.osd_num = 0;
-    msgr.tfd = tfd;
-    msgr.ringloop = ringloop;
-    msgr.repeer_pgs = [this](osd_num_t peer_osd)
+    msgr = new osd_messenger_t();
+    msgr->osd_num = 0;
+    msgr->tfd = tfd;
+    msgr->ringloop = ringloop;
+    msgr->repeer_pgs = [this](osd_num_t peer_osd)
     {
-        if (msgr.osd_peers.find(peer_osd) != msgr.osd_peers.end())
+        if (msgr->osd_peers.find(peer_osd) != msgr->osd_peers.end())
         {
             // peer_osd just connected
             continue_ops();
@@ -45,14 +52,15 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
             }
         }
     };
-    msgr.exec_op = [this](osd_op_t *op)
+    msgr->exec_op = [this](osd_op_t *op)
     {
         // Garbage in
         fprintf(stderr, "Can't handle incoming operation from client %ju\n", op->client_id);
-        msgr.stop_client(op->client_id);
+        msgr->stop_client(op->client_id);
         delete op;
     };
-    msgr.parse_config(config, true);
+    msgr->on_fatal_error_hook = on_fatal_error_hook;
+    msgr->parse_config(config, true);
 
     st_cli = std::move(st_cli_ptr);
     st_cli->on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -64,6 +72,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
     st_cli->on_load_pgs_hook = [this](bool success) { on_load_pgs_hook(success); };
     st_cli->on_reload_hook = [this]() { this->st_cli->load_global_config(); };
     st_cli->on_inode_change_hook = [this](uint64_t inode, bool removed) { on_change_inode_hook(inode, removed); };
+    st_cli->on_fatal_error_hook = on_fatal_error_hook;
 
     st_cli->parse_config(config);
     st_cli->infinite_start = false;
@@ -72,11 +81,26 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
         st_cli->infinite_start = config["client_infinite_start"].bool_value();
     }
     st_cli->load_global_config();
+
+    if (ringloop)
+    {
+        ringloop->loop();
+    }
+
+    msgr->on_fatal_error_hook = st_cli->on_fatal_error_hook = [this](int errcode, const std::string& error)
+    {
+        fatal_error(errcode, error);
+    };
+}
+
+void cluster_client_t::set_on_fatal_error_hook(std::function<void(int, const std::string&)> cb)
+{
+    on_fatal_error_hook = cb;
 }
 
 cluster_client_t::~cluster_client_t()
 {
-    vault_destroy();
+    aborted = true;
     if (retry_timeout_id >= 0)
     {
         tfd->clear_timer(retry_timeout_id);
@@ -89,13 +113,40 @@ cluster_client_t::~cluster_client_t()
         list_retry_timeout_id = -1;
         list_retry_time = {};
     }
-    msgr.repeer_pgs = [](osd_num_t){};
+    msgr->repeer_pgs = [](osd_num_t){};
     if (ringloop)
     {
         ringloop->unregister_consumer(&consumer);
     }
+    delete msgr;
+    msgr = NULL;
+    while (op_queue_head)
+    {
+        op_queue_head->retval = -ECANCELED;
+        erase_op(op_queue_head);
+    }
     delete wb;
     wb = NULL;
+    vault_destroy();
+}
+
+void cluster_client_t::fatal_error(int errcode, const std::string & error)
+{
+    if (on_fatal_error_hook)
+    {
+        if (!aborted)
+        {
+            aborted = true;
+            (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+        }
+        else
+            fprintf(stderr, "Error: %s\n", error.c_str());
+    }
+    else
+    {
+        fprintf(stderr, "Error: %s\n", error.c_str());
+        exit(1);
+    }
 }
 
 cluster_op_t::~cluster_op_t()
@@ -150,13 +201,15 @@ std::vector<osd_num_t> cluster_op_t::get_left_on_dead()
 
 void cluster_client_t::continue_raw_ops(osd_num_t peer_osd)
 {
+    if (aborted)
+        return;
     auto it = raw_ops.find(peer_osd);
     while (it != raw_ops.end() && it->first == peer_osd)
     {
         auto op = it->second;
         op->op_type = OSD_OP_OUT;
-        op->client_id = msgr.osd_peers.at(peer_osd)->client_id;
-        msgr.outbox_push(op);
+        op->client_id = msgr->osd_peers.at(peer_osd)->client_id;
+        msgr->outbox_push(op);
         raw_ops.erase(it++);
     }
 }
@@ -165,14 +218,14 @@ void cluster_client_t::init_msgr()
 {
     if (msgr_initialized)
         return;
-    msgr.init();
+    msgr->init();
     msgr_initialized = true;
     if (ringloop)
     {
         consumer.loop = [this]()
         {
-            msgr.read_requests();
-            msgr.send_replies();
+            msgr->read_requests();
+            msgr->send_replies();
             this->ringloop->submit();
         };
         ringloop->register_consumer(&consumer);
@@ -482,7 +535,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
     }
     // vault
     vault_parse_config();
-    msgr.parse_config(config, false);
+    msgr->parse_config(config, false);
     st_cli->parse_config(config);
     st_cli->load_pgs();
 }
@@ -636,9 +689,9 @@ bool cluster_client_t::get_immediate_commit(uint64_t inode)
 void cluster_client_t::on_change_osd_state_hook(osd_num_t peer_osd)
 {
     osd_tree_metrics.erase(peer_osd);
-    if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
+    if (msgr->wanted_peers.find(peer_osd) != msgr->wanted_peers.end())
     {
-        msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+        msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
         continue_lists();
     }
 }
@@ -855,6 +908,13 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
 
 void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
 {
+    if (aborted)
+    {
+        op->retval = -ECANCELED;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return;
+    }
     if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
     {
         // Atomic writes to multiple stripes are unsupported
@@ -871,7 +931,14 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
         sync_op->flags = 0;
         sync_op->callback = [this, op](cluster_op_t* sync_op)
         {
-            execute_cas(op, true);
+            if (sync_op->retval != 0)
+            {
+                op->retval = sync_op->retval;
+                auto cb = std::move(op->callback);
+                cb(op);
+            }
+            else
+                execute_cas(op, true);
             delete sync_op;
         };
         execute_internal(sync_op);
@@ -887,18 +954,25 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
     op->needs_reslice = false;
     int res = try_send(op, 0, [this, op](osd_op_t *part)
     {
+        if (aborted)
+        {
+            op->retval = -ECANCELED;
+            auto cb = std::move(op->callback);
+            cb(op);
+            return;
+        }
         int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len;
         op->retval = part->reply.hdr.retval;
         if (op->retval != expected && op->retval >= 0)
             op->retval = -EIO;
         op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
-        auto peer_it = msgr.osd_peers.find(op->parts[0].osd_num);
+        auto peer_it = msgr->osd_peers.find(op->parts[0].osd_num);
         if (op->retval != expected || (op->flags & OP_IMMEDIATE_COMMIT))
         {
             auto cb = std::move(op->callback);
             cb(op);
         }
-        else if (peer_it == msgr.osd_peers.end())
+        else if (peer_it == msgr->osd_peers.end())
         {
             // Care must be taken to make sure that the client doesn't reconnect to the OSD
             // before executing the previously completed operation callback (!)
@@ -930,7 +1004,7 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
                     cb(op);
                 },
             };
-            msgr.outbox_push(part);
+            msgr->outbox_push(part);
         }
     });
     if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE)
@@ -1061,23 +1135,25 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
 
 void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op)
 {
-    auto peer_it = msgr.osd_peers.find(osd_num);
-    if (peer_it != msgr.osd_peers.end())
+    auto peer_it = msgr->osd_peers.find(osd_num);
+    if (peer_it != msgr->osd_peers.end())
     {
         op->op_type = OSD_OP_OUT;
         op->client_id = peer_it->second->client_id;
-        msgr.outbox_push(op);
+        msgr->outbox_push(op);
     }
     else
     {
-        if (msgr.wanted_peers.find(osd_num) == msgr.wanted_peers.end())
-            msgr.connect_peer(osd_num, st_cli->peer_states[osd_num]);
+        if (msgr->wanted_peers.find(osd_num) == msgr->wanted_peers.end())
+            msgr->connect_peer(osd_num, st_cli->peer_states[osd_num]);
         raw_ops.emplace(osd_num, op);
     }
 }
 
 int cluster_client_t::continue_rw(cluster_op_t *op)
 {
+    if (aborted)
+        return 0;
     if (op->state == 0)
         goto resume_0;
     else if (op->state == 1)
@@ -1450,8 +1526,8 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                 primary_osd = nearest_osd;
         }
         part->osd_num = primary_osd;
-        auto peer_it = msgr.osd_peers.find(primary_osd);
-        if (peer_it != msgr.osd_peers.end())
+        auto peer_it = msgr->osd_peers.find(primary_osd);
+        if (peer_it != msgr->osd_peers.end())
         {
             osd_client_t *cl = peer_it->second;
             part->flags |= PART_SENT|PART_VALID;
@@ -1495,12 +1571,12 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                 },
             };
             part->op.iov = part->iov;
-            msgr.outbox_push(&part->op);
+            msgr->outbox_push(&part->op);
             return TRY_SEND_OK;
         }
-        else if (msgr.wanted_peers.find(primary_osd) == msgr.wanted_peers.end())
+        else if (msgr->wanted_peers.find(primary_osd) == msgr->wanted_peers.end())
         {
-            msgr.connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
+            msgr->connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
             return TRY_SEND_CONNECTING;
         }
     }
@@ -1509,6 +1585,8 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
 
 int cluster_client_t::continue_sync(cluster_op_t *op)
 {
+    if (aborted)
+        return 0;
     if (op->state == 1)
         goto resume_1;
     if (!dirty_osds.size())
@@ -1522,8 +1600,8 @@ int cluster_client_t::continue_sync(cluster_op_t *op)
     for (auto do_it = dirty_osds.begin(); do_it != dirty_osds.end(); )
     {
         osd_num_t sync_osd = *do_it;
-        auto peer_it = msgr.osd_peers.find(sync_osd);
-        if (peer_it == msgr.osd_peers.end())
+        auto peer_it = msgr->osd_peers.find(sync_osd);
+        if (peer_it == msgr->osd_peers.end())
             dirty_osds.erase(do_it++);
         else
             do_it++;
@@ -1578,7 +1656,7 @@ resume_1:
 
 void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
 {
-    osd_client_t *cl = msgr.osd_peers.at(part->osd_num);
+    osd_client_t *cl = msgr->osd_peers.at(part->osd_num);
     part->flags |= PART_SENT;
     op->inflight_count++;
     part->op = (osd_op_t){
@@ -1595,7 +1673,7 @@ void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
             handle_op_part(part);
         },
     };
-    msgr.outbox_push(&part->op);
+    msgr->outbox_push(&part->op);
 }
 
 static inline void mem_or(void *res, const void *r2, unsigned int len)
@@ -1650,7 +1728,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         reset_retry_timer(op->retry_after);
         if (stop_client_id)
         {
-            msgr.stop_client(stop_client_id);
+            msgr->stop_client(stop_client_id);
         }
         op->inflight_count--;
         if (op->inflight_count == 0 && !op->retry_after)

src/client/cluster_client.h

@@ -110,6 +110,7 @@ class __attribute__((visibility("default"))) cluster_client_t
 {
     timerfd_manager_t *tfd = NULL;
     ring_loop_t *ringloop = NULL;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
 
     // config:
 
@@ -178,11 +179,12 @@ class __attribute__((visibility("default"))) cluster_client_t
     std::multimap<osd_num_t, osd_op_t*> raw_ops;
     int continuing_ops = 0;
     bool msgr_initialized = false;
+    bool aborted = false;
 
 public:
     std::unique_ptr<etcd_state_client_t> st_cli;
+    osd_messenger_t* msgr = NULL;
 
-    osd_messenger_t msgr;
     void init_msgr();
 
     json11::Json::object cli_config, file_config, etcd_global_config;
@@ -196,6 +198,7 @@ public:
     bool is_ready();
     void on_ready(std::function<void(void)> fn);
     bool flush();
+    void set_on_fatal_error_hook(std::function<void(int, const std::string&)>);
 
     bool get_immediate_commit(uint64_t inode);
 
@@ -204,6 +207,7 @@ public:
 
 protected:
     void continue_ops(int time_passed = 0);
+    void fatal_error(int errcode, const std::string & error);
 
     std::shared_ptr<inode_cache_t> inode_cache_get(inode_t ino);
     void vault_parse_config();

src/client/cluster_client_impl.h

@@ -46,6 +46,7 @@ public:
     int writeback_queue_size = 0;
     int writebacks_active = 0;
     uint64_t last_flush_id = 0;
+    cluster_client_t *parent = NULL;
 
     std::map<object_id, cluster_buffer_t> dirty_buffers;
     std::vector<cluster_op_t*> writeback_overflow;

src/client/cluster_client_list.cpp

@@ -301,14 +301,14 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg)
     bool conn = true;
     for (osd_num_t peer_osd: all_peers)
     {
-        if (msgr.osd_peers.find(peer_osd) == msgr.osd_peers.end())
+        if (msgr->osd_peers.find(peer_osd) == msgr->osd_peers.end())
         {
             // Initiate connection
             if (st_cli->peer_states[peer_osd].is_null())
             {
                 return LIST_PG_WAIT_ACTIVE;
             }
-            msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+            msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
             conn = false;
         }
     }
@@ -346,7 +346,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
     osd_op_t *op = new osd_op_t();
     op->op_type = OSD_OP_OUT;
     // Already checked that it exists above, but anyway
-    op->client_id = msgr.osd_peers.at(cur_list->osd_num)->client_id;
+    op->client_id = msgr->osd_peers.at(cur_list->osd_num)->client_id;
     op->req = (osd_any_op_t){
         .sec_list = {
             .header = {
@@ -425,7 +425,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
         finish_list_pg(cur_list->pg, true);
         continue_listing(cur_list->pg->lst);
     };
-    msgr.outbox_push(op);
+    msgr->outbox_push(op);
 }
 
 void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)

src/client/cluster_client_real.cpp

@@ -98,8 +98,8 @@ void cluster_client_t::vault_load_keys()
         vault_http_ctx = http_context_init(tfd, vault_client_cert, vault_client_key, vault_ca, true, error);
         if (!vault_http_ctx)
         {
-            fprintf(stderr, "Failed to initialize HTTP context for Vault: %s\n", error.c_str());
-            exit(1);
+            fatal_error(EINVAL, "Failed to initialize HTTP context for Vault: "+error);
+            return;
         }
         vault_http_cli = http_init(vault_http_ctx);
     }

src/client/cluster_client_wb.cpp

@@ -336,7 +336,7 @@ void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from
     {
         // Buffer flushes are always retried, regardless of the error,
         // so they should never result in an error here
-        assert(op->retval == op->len);
+        assert(op->retval == op->len || parent->aborted);
         for (auto fl_it = flushed_buffers.lower_bound(op->flush_id);
             fl_it != flushed_buffers.end() && fl_it->first == op->flush_id; )
         {

src/client/etcd_state_client.cpp

@@ -10,6 +10,19 @@
 #include "str_util.h"
 #include "json_util.h"
 
+void etcd_state_client_t::fatal_error(int errcode, const std::string & error)
+{
+    if (on_fatal_error_hook)
+    {
+        (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+    }
+    else
+    {
+        fprintf(stderr, "Error: %s\n", error.c_str());
+        exit(1);
+    }
+}
+
 etcd_state_client_t::~etcd_state_client_t()
 {
     for (auto watch: watches)

src/client/etcd_state_client.h

@@ -144,6 +144,7 @@ protected:
     std::set<osd_num_t> seen_peers;
     bool new_pg_config = false;
 
+    void fatal_error(int errcode, const std::string & error);
     void add_etcd_url(std::string);
     void reset_pg_exists();
     void clean_nonexistent_pgs();
@@ -192,6 +193,7 @@ public:
     std::function<void()> on_reload_hook;
     std::function<void(inode_t, bool)> on_inode_change_hook;
     std::function<void(http_co_t *)> on_start_watcher_hook;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
 
     json11::Json::object serialize_inode_cfg(inode_config_t *cfg);
     inode_config_t deserialize_inode_cfg(uint64_t inode_num, json11::Json value, uint64_t mod_revision);

src/client/etcd_state_client_http.cpp

@@ -49,7 +49,11 @@ void etcd_state_client_http_t::etcd_add_watch(json11::Json watch)
 
 std::string etcd_state_client_http_t::get_username()
 {
-    return http_context_get_ssl_cn(get_http_ctx());
+    if (!get_http_ctx())
+    {
+        return "";
+    }
+    return http_context_get_ssl_cn(http_ctx);
 }
 
 http_context_t *etcd_state_client_http_t::get_http_ctx()
@@ -60,8 +64,7 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
         http_ctx = http_context_init(tfd, etcd_client_cert, etcd_client_key, etcd_ca, true, error);
         if (!http_ctx)
         {
-            fprintf(stderr, "Failed to initialize HTTP context: %s\n", error.c_str());
-            exit(1);
+            fatal_error(EINVAL, "Failed to initialize HTTP context: "+error);
         }
     }
     return http_ctx;
@@ -70,7 +73,11 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
 void etcd_state_client_http_t::etcd_call_oneshot(const std::string & etcd_url, const std::string & api, json11::Json payload,
     int timeout, std::function<void(std::string, json11::Json)> callback)
 {
-    auto http_cli = http_init(get_http_ctx());
+    if (!get_http_ctx())
+    {
+        return;
+    }
+    auto http_cli = http_init(http_ctx);
     http_json_post(http_cli, etcd_url+api, payload, "", { .timeout = timeout }, [http_cli, callback](http_message_t *response)
     {
         std::string err;
@@ -140,7 +147,11 @@ void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json1
             callback(err, data);
     };
     if (!keepalive_client)
-        keepalive_client = http_init(get_http_ctx());
+    {
+        if (!get_http_ctx())
+            return;
+        keepalive_client = http_init(http_ctx);
+    }
     http_request(keepalive_client, url.addr, req, { .timeout = timeout, .keepalive = true, .ssl = url.ssl }, cb);
 }
 
@@ -159,8 +170,12 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
 {
     if (!etcd_addresses.size() && !etcd_local.size())
     {
-        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
-        exit(1);
+        fatal_error(EINVAL, "etcd_address is missing in Vitastor configuration");
+        return;
+    }
+    if (!get_http_ctx())
+    {
+        return;
     }
     if (selected_etcd_url.addr != "")
     {
@@ -196,7 +211,7 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
     for (auto & url: etcd_name_urls)
     {
         resolve_count++;
-        http_resolve(get_http_ctx(), url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
+        http_resolve(http_ctx, url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
         {
             if (error != "")
                 fprintf(stderr, "Error resolving %s: %s\n", url.addr.c_str(), error.c_str());
@@ -228,8 +243,8 @@ void etcd_state_client_http_t::pick_next_etcd_on_resolve()
 {
     if (!etcd_urls_to_try.size())
     {
-        fprintf(stderr, "None of etcd_address could be resolved\n");
-        exit(1);
+        fatal_error(EIO, "None of etcd_address could be resolved");
+        return;
     }
     if (!rand_initialized)
     {
@@ -274,7 +289,11 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
             etcd_watch_revision_config, etcd_watch_revision_osd, etcd_watch_revision_pg);
     }
     if (!etcd_watch_ws)
-        etcd_watch_ws = http_init(get_http_ctx());
+    {
+        if (!get_http_ctx())
+            return;
+        etcd_watch_ws = http_init(http_ctx);
+    }
     else
         http_close(etcd_watch_ws);
     open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl },
@@ -324,20 +343,17 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
                                 etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = 0;
                                 on_reload_hook();
                             }
-                            return;
                         }
                         else
                         {
-                            fprintf(stderr, "Revisions before %ju were compacted by etcd, exiting\n",
-                                data["result"]["compact_revision"].uint64_value());
-                            exit(1);
+                            fatal_error(EIO, "Revisions before "+data["result"]["compact_revision"].as_string()+" were compacted by etcd, exiting");
                         }
                     }
                     else
                     {
-                        fprintf(stderr, "Watch canceled by etcd, reason: %s, exiting\n", data["result"]["cancel_reason"].string_value().c_str());
-                        exit(1);
+                        fatal_error(EIO, "Watch canceled by etcd, reason: "+data["result"]["cancel_reason"].string_value()+", exiting");
                     }
+                    return;
                 }
                 // Save revision only if it's present in the message - because sometimes etcd sends something without a header, like:
                 // {"error": {"grpc_code": 14, "http_code": 503, "http_status": "Service Unavailable", "message": "error reading from server: EOF"}}
@@ -492,9 +508,9 @@ void etcd_state_client_http_t::load_global_config()
     {
         if (err != "")
         {
-            fprintf(stderr, "Error reading configuration from etcd: %s\n", err.c_str());
             if (infinite_start)
             {
+                fprintf(stderr, "Failed to read configuration from etcd: %s\n", err.c_str());
                 tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
                 {
                     load_global_config();
@@ -502,7 +518,7 @@ void etcd_state_client_http_t::load_global_config()
             }
             else
             {
-                exit(1);
+                fatal_error(EIO, "Failed to read configuration from etcd: "+err);
             }
         }
     });

src/client/messenger.cpp

@@ -17,6 +17,19 @@
 #include "msgr_rdma.h"
 #endif
 
+void osd_messenger_t::fatal_error(int errcode, const std::string & error)
+{
+    if (on_fatal_error_hook)
+    {
+        (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+    }
+    else
+    {
+        fprintf(stderr, "Error: %s\n", error.c_str());
+        exit(1);
+    }
+}
+
 void osd_messenger_t::init()
 {
     init_tls();
@@ -201,7 +214,7 @@ osd_messenger_t::~osd_messenger_t()
     destroy_tls();
 }
 
-static int parse_proto_checksums(const json11::Json & val, int default_value)
+int osd_messenger_t::parse_proto_checksums(const json11::Json & val, int default_value)
 {
     if (val.is_string())
     {
@@ -219,9 +232,9 @@ static int parse_proto_checksums(const json11::Json & val, int default_value)
     }
     else if (val.is_null())
         return default_value;
-    fprintf(stderr, "proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
-        ", \"\" or null (default), but it is: %s\n", val.dump().c_str());
-    exit(1);
+    fatal_error(EINVAL, "proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
+        ", \"\" or null (default), but it is: "+val.dump());
+    return default_value;
 }
 
 void osd_messenger_t::parse_config(const json11::Json & config, bool init)
@@ -813,7 +826,7 @@ void osd_messenger_t::accept_connections(int listen_fd)
     }
     if (peer_fd == -1 && errno != EAGAIN)
     {
-        throw std::runtime_error(std::string("accept: ") + strerror(errno));
+        fatal_error(EIO, std::string("accept: ") + strerror(errno));
     }
 }
 

src/client/messenger.h

@@ -259,6 +259,7 @@ protected:
     bool gcm_enabled = false;
     msgr_handshake_ctx_i *hs_ctx = NULL;
 
+    void fatal_error(int errcode, const std::string & error);
     void init_tls();
     void destroy_tls();
     bool init_tls_client(osd_client_t *cl);
@@ -319,6 +320,7 @@ public:
     std::function<void(osd_num_t)> break_pg_locks;
     std::function<bool(osd_client_t*, json11::Json)> check_config_hook;
     std::function<void(osd_client_t*)> handshake_hook;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
     void read_requests();
     void send_replies();
     void accept_connections(int listen_fd);
@@ -346,7 +348,8 @@ public:
     void measure_exec(osd_op_t *cur_op);
 
 protected:
-    void try_connect_peer(osd_num_t osd_num);
+    int parse_proto_checksums(const json11::Json & val, int default_value);
+    void try_connect_peer(uint64_t osd_num);
     void try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_host, int peer_port);
     void handle_peer_epoll(int peer_fd, int epoll_events);
     void handle_connect_epoll(int peer_fd);

src/client/msgr_encrypt.cpp

@@ -597,10 +597,9 @@ void osd_messenger_t::init_tls()
         if (client_cert.empty() || client_key.empty() || osd_ca.empty() || osd_num && client_ca.empty())
         {
             if (osd_num)
-                fprintf(stderr, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs\n");
+                fatal_error(EINVAL, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs");
             else
-                fprintf(stderr, "Vitastor transport encryption requires cert, pkey and osd_ca options\n");
-            exit(1);
+                fatal_error(EINVAL, "Vitastor transport encryption requires cert, pkey and osd_ca options");
         }
         else
         {
@@ -609,8 +608,8 @@ void osd_messenger_t::init_tls()
             hs_ctx = msgr_handshake_ctx_i::create_ctx();
             if (!hs_ctx->init(client_cert, client_key, osd_ca, client_ca, admin_ca))
             {
-                fprintf(stderr, "Error: %s\n", hs_ctx->get_error().c_str());
-                exit(1);
+                gcm_enabled = false;
+                fatal_error(EINVAL, hs_ctx->get_error());
             }
 #endif
         }

src/client/msgr_rdma.cpp

@@ -575,7 +575,7 @@ json11::Json osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_
     return json11::Json();
 }
 
-static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
+static int try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
 {
     ibv_send_wr *bad_wr = NULL;
     ibv_send_wr wr = {
@@ -586,12 +586,12 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
         .send_flags = IBV_SEND_SIGNALED,
     };
     int err = ibv_post_send(cl->rdma_conn->qp, &wr, &bad_wr);
-    if (err || bad_wr)
-    {
-        fprintf(stderr, "RDMA send failed: %s (code %d)\n", strerror(err), err);
-        exit(1);
-    }
+    if (err)
+        return err;
+    if (bad_wr)
+        return EIO;
     cl->rdma_conn->cur_send++;
+    return 0;
 }
 
 bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
@@ -605,8 +605,11 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
         rc->send_out.mr = ibv_reg_mr(rc->ctx->pd, rc->send_out.buf, rc->send_out_size, 0);
         if (!rc->send_out.mr)
         {
-            fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-            exit(1);
+            rc->send_out_size = 0;
+            free(rc->send_out.buf);
+            rc->send_out.buf = NULL;
+            fatal_error(EIO, std::string("Failed to register RDMA memory region: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return false;
         }
     }
     // Copy data into the buffer and send it
@@ -641,7 +644,12 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
                 .length = (uint32_t)copied,
                 .lkey = rc->send_out.mr->lkey,
             };
-            try_send_rdma_wr(cl, &sge, 1);
+            int err = try_send_rdma_wr(cl, &sge, 1);
+            if (err)
+            {
+                fatal_error(EIO, std::string("RDMA send failed: ")+strerror(err)+" (code "+std::to_string(err)+")");
+                return false;
+            }
             rc->send_sizes.push_back(copied);
             cl->unref_send_free(send_free_from, cl->next_send_free + cl->send_free.size()); // unref at once - everything is already copied
         }
@@ -649,7 +657,7 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
     return true;
 }
 
-static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
+static int try_recv_rdma_wr(osd_client_t *cl, void *buf)
 {
     ibv_sge sge = {
         .addr = (uintptr_t)buf,
@@ -663,12 +671,12 @@ static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
         .num_sge = 1,
     };
     int err = ibv_post_recv(cl->rdma_conn->qp, &wr, &bad_wr);
-    if (err || bad_wr)
-    {
-        fprintf(stderr, "RDMA receive failed: %s\n", strerror(err));
-        exit(1);
-    }
+    if (err)
+        return err;
+    if (bad_wr)
+        return EIO;
     cl->rdma_conn->cur_recv++;
+    return 0;
 }
 
 bool osd_messenger_t::init_recv_rdma(osd_client_t *cl)
@@ -679,14 +687,21 @@ bool osd_messenger_t::init_recv_rdma(osd_client_t *cl)
     rc->recv_buf.mr = ibv_reg_mr(rc->ctx->pd, rc->recv_buf.buf, rc->max_msg * rc->max_recv, IBV_ACCESS_LOCAL_WRITE);
     if (!rc->recv_buf.mr)
     {
-        fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-        exit(1);
+        free(rc->recv_buf.buf);
+        rc->recv_buf.buf = NULL;
+        fatal_error(EIO, std::string("Failed to register RDMA memory region: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+        return false;
     }
     for (uint32_t i = 0; i < rc->max_recv; i++)
     {
         uint8_t *b = rc->recv_buf.buf + i*rc->max_msg;
         rc->recv_buffers.push_back(b);
-        try_recv_rdma_wr(cl, b);
+        int err = try_recv_rdma_wr(cl, b);
+        if (err)
+        {
+            fatal_error(EIO, std::string("RDMA receive failed: ")+strerror(err)+" (code "+std::to_string(err)+")");
+            return false;
+        }
     }
     return true;
 }
@@ -705,8 +720,8 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
     }
     if (ibv_req_notify_cq(rdma_context->cq, 0) != 0)
     {
-        fprintf(stderr, "Failed to request RDMA completion notification, exiting\n");
-        exit(1);
+        fatal_error(EIO, "Failed to request RDMA completion notification");
+        return;
     }
     ibv_wc wc[RDMA_EVENTS_AT_ONCE];
     int event_count;

src/client/msgr_rdmacm.cpp

@@ -92,8 +92,8 @@ void osd_messenger_t::handle_rdmacm_events()
         {
             if (errno == EAGAIN || errno == EINTR)
                 break;
-            fprintf(stderr, "Failed to get RDMA-CM event: %s (code %d)\n", strerror(errno), errno);
-            exit(1);
+            fatal_error(EIO, std::string("Failed to get RDMA-CM event: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return;
         }
         // ...so we save a copy of all events EXCEPT connection requests, otherwise they sometimes fail with EVENT_DISCONNECT
         if (ev->event == RDMA_CM_EVENT_CONNECT_REQUEST)
@@ -107,8 +107,8 @@ void osd_messenger_t::handle_rdmacm_events()
         r = rdma_ack_cm_event(ev);
         if (r != 0)
         {
-            fprintf(stderr, "Failed to ack (free) RDMA-CM event: %s (code %d)\n", strerror(errno), errno);
-            exit(1);
+            fatal_error(EIO, std::string("Failed to ack (free) RDMA-CM event: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return;
         }
     }
     for (auto & evl: events_copy)

src/client/vitastor_c.cpp

@@ -30,6 +30,18 @@ struct vitastor_c
 
     QEMUSetFDHandler *aio_set_fd_handler = NULL;
     void *aio_ctx = NULL;
+
+    ~vitastor_c()
+    {
+        if (cli)
+            delete cli;
+        if (epmgr)
+            delete epmgr;
+        else if (tfd)
+            delete tfd;
+        if (ringloop)
+            delete ringloop;
+    }
 };
 
 extern "C" {
@@ -103,7 +115,16 @@ vitastor_c *vitastor_c_create_qemu(QEMUSetFDHandler *aio_set_fd_handler, void *a
         rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
     );
     auto self = vitastor_c_create_qemu_common(aio_set_fd_handler, aio_context);
-    self->cli = cluster_client_t::create(NULL, self->tfd, cfg_json);
+    try
+    {
+        self->cli = cluster_client_t::create(NULL, self->tfd, cfg_json);
+    }
+    catch (std::exception & e)
+    {
+        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
+        delete self;
+        return NULL;
+    }
     return self;
 }
 
@@ -111,47 +132,45 @@ vitastor_c *vitastor_c_create_qemu_uring(QEMUSetFDHandler *aio_set_fd_handler, v
     const char *config_path, const char *etcd_host, const char *etcd_prefix,
     int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level)
 {
-    ring_loop_t *ringloop = NULL;
-    try
-    {
-        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-    }
-    catch (std::exception & e)
-    {
-        return NULL;
-    }
     json11::Json cfg_json = vitastor_c_common_config(
         config_path, etcd_host, etcd_prefix, use_rdma,
         rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
     );
     auto self = vitastor_c_create_qemu_common(aio_set_fd_handler, aio_context);
-    self->ringloop = ringloop;
-    self->cli = cluster_client_t::create(self->ringloop, self->tfd, cfg_json);
-    ringloop->loop();
+    try
+    {
+        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+        self->cli = cluster_client_t::create(self->ringloop, self->tfd, cfg_json);
+    }
+    catch (std::exception & e)
+    {
+        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
+        delete self;
+        return NULL;
+    }
     return self;
 }
 
 vitastor_c *vitastor_c_create_uring(const char *config_path, const char *etcd_host, const char *etcd_prefix,
     int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level)
 {
-    ring_loop_t *ringloop = NULL;
-    try
-    {
-        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-    }
-    catch (std::exception & e)
-    {
-        return NULL;
-    }
     json11::Json cfg_json = vitastor_c_common_config(
         config_path, etcd_host, etcd_prefix, use_rdma,
         rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
     );
     vitastor_c *self = new vitastor_c;
-    self->ringloop = ringloop;
-    self->epmgr = new epoll_manager_t(self->ringloop);
-    self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
-    ringloop->loop();
+    try
+    {
+        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+        self->epmgr = new epoll_manager_t(self->ringloop);
+        self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
+    }
+    catch (std::exception & e)
+    {
+        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
+        delete self;
+        return NULL;
+    }
     return self;
 }
 
@@ -166,15 +185,6 @@ int vitastor_c_uring_register_eventfd(vitastor_c *client)
 
 vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
 {
-    ring_loop_t *ringloop = NULL;
-    try
-    {
-        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-    }
-    catch (std::exception & e)
-    {
-        return NULL;
-    }
     json11::Json::object cfg;
     for (int i = 0; i < options_len-1; i += 2)
     {
@@ -189,10 +199,18 @@ vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
     }
     json11::Json cfg_json(cfg);
     vitastor_c *self = new vitastor_c;
-    self->ringloop = ringloop;
-    self->epmgr = new epoll_manager_t(self->ringloop);
-    self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
-    ringloop->loop();
+    try
+    {
+        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+        self->epmgr = new epoll_manager_t(self->ringloop);
+        self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
+    }
+    catch (std::exception & e)
+    {
+        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
+        delete self;
+        return NULL;
+    }
     return self;
 }
 
@@ -205,11 +223,28 @@ vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len)
     }
     json11::Json cfg_json(cfg);
     vitastor_c *self = new vitastor_c;
-    self->epmgr = new epoll_manager_t(NULL);
-    self->cli = cluster_client_t::create(NULL, self->epmgr->tfd, cfg_json);
+    try
+    {
+        self->epmgr = new epoll_manager_t(NULL);
+        self->cli = cluster_client_t::create(NULL, self->epmgr->tfd, cfg_json);
+    }
+    catch (std::exception & e)
+    {
+        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
+        delete self;
+        return NULL;
+    }
     return self;
 }
 
+void vitastor_c_on_fatal_error(vitastor_c *client, VitastorErrorHandler err, void *opaque)
+{
+    client->cli->set_on_fatal_error_hook([err, opaque](int retval, const std::string& errstr)
+    {
+        err(opaque, retval, errstr.c_str());
+    });
+}
+
 void* vitastor_c_get_internal_client(vitastor_c *client)
 {
     return client->cli;
@@ -217,13 +252,6 @@ void* vitastor_c_get_internal_client(vitastor_c *client)
 
 void vitastor_c_destroy(vitastor_c *client)
 {
-    delete client->cli;
-    if (client->epmgr)
-        delete client->epmgr;
-    else if (client->tfd)
-        delete client->tfd;
-    if (client->ringloop)
-        delete client->ringloop;
     delete client;
 }
 

src/client/vitastor_c.h

@@ -34,6 +34,7 @@ typedef struct vitastor_c vitastor_c;
 typedef void VitastorReadHandler(void *opaque, long retval, uint64_t version);
 typedef void VitastorIOHandler(void *opaque, long retval);
 typedef void VitastorReadBitmapHandler(void *opaque, long retval, uint8_t *bitmap);
+typedef void VitastorErrorHandler(void *opaque, long retval, const char* errstr);
 
 // QEMU
 typedef void IOHandler(void *opaque);
@@ -50,6 +51,7 @@ vitastor_c *vitastor_c_create_uring(const char *config_path, const char *etcd_ho
     int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level);
 vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len);
 vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len);
+void vitastor_c_on_fatal_error(vitastor_c *client, VitastorErrorHandler err, void *opaque);
 void* vitastor_c_get_internal_client(vitastor_c *client);
 void vitastor_c_destroy(vitastor_c *client);
 int vitastor_c_is_ready(vitastor_c *client);

src/test/mock/messenger.cpp

@@ -8,6 +8,12 @@
 
 #include "messenger.h"
 
+void osd_messenger_t::fatal_error(int retval, const std::string & error)
+{
+    fprintf(stderr, "%s\n", error.c_str());
+    exit(1);
+}
+
 void osd_messenger_t::init()
 {
 }

src/test/test_cluster_client.cpp

@@ -168,16 +168,16 @@ void pretend_disconnected(osd_messenger_t *msgr, osd_num_t osd_num)
 
 void pretend_connected(cluster_client_t *cli, osd_num_t osd_num)
 {
-    pretend_connected(&cli->msgr, osd_num);
+    pretend_connected(cli->msgr, osd_num);
 }
 
 void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 {
-    pretend_disconnected(&cli->msgr, osd_num);
+    pretend_disconnected(cli->msgr, osd_num);
 }
 
 #define check_disconnected(cli, osd_num) {\
-    if (cli->msgr.osd_peers.find(osd_num) != cli->msgr.osd_peers.end())\
+    if (cli->msgr->osd_peers.find(osd_num) != cli->msgr->osd_peers.end())\
     {\
         printf("OSD %ju not disconnected as it ought to be\n", (uint64_t)osd_num);\
         assert(0);\
@@ -185,7 +185,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 }
 
 #define check_op_count(cli, osd_num, ops) {\
-    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);\
+    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);\
     int real_ops = cl->sent_ops.size();\
     if (real_ops != ops)\
     {\
@@ -196,7 +196,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 
 osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uint64_t offset, uint64_t len)
 {
-    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);
+    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);
     auto op_it = cl->sent_ops.begin();
     while (op_it != cl->sent_ops.end())
     {
@@ -225,7 +225,7 @@ void pretend_op_completed(cluster_client_t *cli, osd_op_t *op, int64_t retval)
         ? "sync" : (op->req.hdr.opcode == OSD_OP_WRITE ? "write" : "read"), op->req.rw.offset, op->req.rw.len);
     uint64_t op_id = op->req.hdr.id;
     uint64_t client_id = op->client_id;
-    cli->msgr.clients[client_id]->sent_ops.erase(op_id);
+    cli->msgr->clients[client_id]->sent_ops.erase(op_id);
     op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
     op->reply.hdr.id = op->req.hdr.id;
     op->reply.hdr.opcode = op->req.hdr.opcode;
@@ -309,7 +309,7 @@ void test1()
         uint64_t replay_start = UINT64_MAX;
         uint64_t replay_end = 0;
         std::vector<osd_op_t*> replay_ops;
-        auto osd_cl = cli->msgr.osd_peers.at(1);
+        auto osd_cl = cli->msgr->osd_peers.at(1);
         for (auto & op_p: osd_cl->sent_ops)
         {
             auto op = op_p.second;
@@ -742,7 +742,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *read_op = cl->sent_ops.begin()->second;
         assert(read_op->req.hdr.opcode == OSD_OP_READ);
         assert(read_op->req.rw.inode == child_inode);
@@ -784,7 +784,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *read_op = cl->sent_ops.begin()->second;
         assert(read_op->req.hdr.opcode == OSD_OP_READ);
         assert(read_op->req.rw.inode == child_inode);
@@ -818,7 +818,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *child_read = cl->sent_ops.begin()->second;
         assert(child_read->req.hdr.opcode == OSD_OP_READ);
         assert(child_read->req.rw.inode == child_inode);