return -1;
// start up network
- SimpleMessenger *messenger = new SimpleMessenger(g_ceph_context, entity_name_t::CLIENT());
+ SimpleMessenger *messenger = new SimpleMessenger(g_ceph_context,
+ entity_name_t::CLIENT());
+ messenger->set_nonce(getpid());
Client *client = new Client(messenger, &mc);
if (filer_flags) {
client->set_filer_flags(filer_flags);
::close(fd[0]);
cout << "ceph-fuse[" << getpid() << "]: starting ceph client" << std::endl;
- messenger->start_with_nonce(getpid());
+ messenger->start();
// start client
client->init();
SimpleMessenger *messenger = new SimpleMessenger(g_ceph_context,
entity_name_t::CLIENT());
int r = messenger->bind(g_conf->public_addr, getpid());
+ messenger->set_nonce(getpid());
if (r < 0)
return r;
MonClient mc(g_ceph_context);
SimpleMessenger *client_messenger = new SimpleMessenger(g_ceph_context, entity_name_t::OSD(whoami));
SimpleMessenger *cluster_messenger = new SimpleMessenger(g_ceph_context, entity_name_t::OSD(whoami));
- SimpleMessenger *messenger_hbin = new SimpleMessenger(g_ceph_context, entity_name_t::OSD(whoami));
+ SimpleMessenger *messenger_hbin = new SimpleMessenger(g_ceph_context,entity_name_t::OSD(whoami));
SimpleMessenger *messenger_hbout = new SimpleMessenger(g_ceph_context, entity_name_t::OSD(whoami));
cluster_messenger->set_cluster_protocol(CEPH_OSD_PROTOCOL);
messenger_hbin->set_cluster_protocol(CEPH_OSD_PROTOCOL);
+ messenger_hbin->set_nonce(getpid());
messenger_hbout->set_cluster_protocol(CEPH_OSD_PROTOCOL);
r = client_messenger->bind(g_conf->public_addr, getpid());
global_init_shutdown_stderr(g_ceph_context);
client_messenger->start();
- messenger_hbin->start_with_nonce(getpid());
+ messenger_hbin->start();
messenger_hbout->start();
cluster_messenger->start();
//network connection
messenger = new SimpleMessenger(cct, entity_name_t::CLIENT());
+ messenger->set_nonce(msgr_nonce);
//at last the client
ret = -1002;
goto fail;
ret = -1003;
- if (messenger->start_with_nonce(msgr_nonce) != 0)
+ if (messenger->start() != 0)
goto fail;
ret = client->init();
goto out;
err = -ENOMEM;
+ nonce = getpid() + (1000000 * (uint64_t)rados_instance.inc());
messenger = new SimpleMessenger(cct, entity_name_t::CLIENT(-1));
if (!messenger)
goto out;
+ messenger->set_nonce(nonce);
// require OSDREPLYMUX feature. this means we will fail to talk to
// old servers. this is necessary because otherwise we won't know
// how to decompose the reply data into its consituent pieces.
messenger->add_dispatcher_head(this);
- nonce = getpid() + (1000000 * (uint64_t)rados_instance.inc());
-
- messenger->start_with_nonce(nonce);
+ messenger->start();
messenger->add_dispatcher_head(this);
ldout(cct, 1) << "setting wanted keys" << dendl;
objecter->set_client_incarnation(0);
messenger->add_dispatcher_head(this);
- messenger->start_with_nonce(getpid());
+ messenger->start();
monc->set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD|CEPH_ENTITY_TYPE_MDS);
monc->set_messenger(messenger);
objecter->set_client_incarnation(0);
messenger->add_dispatcher_head(this);
- messenger->start_with_nonce(getpid());
+ messenger->start();
monc->set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD|CEPH_ENTITY_TYPE_MDS);
monc->set_messenger(messenger);
SimpleMessenger* smessenger = NULL;
if (!messenger) {
messenger = smessenger = new SimpleMessenger(cct, entity_name_t::CLIENT(-1));
+ smessenger->set_nonce(getpid());
messenger->add_dispatcher_head(this);
- smessenger->start_with_nonce(getpid());
+ smessenger->start();
temp_msgr = true;
}
return accepter.rebind(avoid_port);
}
-int SimpleMessenger::start_with_nonce(uint64_t nonce)
+int SimpleMessenger::start()
{
lock.Lock();
ldout(cct,1) << "messenger.start" << dendl;
// register at least one entity, first!
- assert(my_type >= 0);
+ assert(my_type >= 0);
assert(!started);
started = true;
// where i listen
bool need_addr;
entity_addr_t ms_addr;
+ uint64_t nonce;
+ void set_nonce(uint64_t new_nonce) { nonce = new_nonce; }
// local
bool destination_stopped;
accepter(this),
lock("SimpleMessenger::lock"), started(false), did_bind(false),
dispatch_throttler(cct->_conf->ms_dispatch_throttle_bytes), need_addr(true),
- destination_stopped(false), my_type(name.type()),
+ nonce(0), destination_stopped(false), my_type(name.type()),
global_seq_lock("SimpleMessenger::global_seq_lock"), global_seq(0),
reaper_thread(this), reaper_started(false), reaper_stop(false),
dispatch_thread(this), msgr(this),
delete dispatch_queue.local_pipe;
}
- //void set_listen_addr(tcpaddr_t& a);
-
int bind(entity_addr_t bind_addr, int64_t nonce);
- int start_with_nonce(uint64_t nonce); // if we didn't bind
- virtual int start() { // if we did
- assert(did_bind);
- return start_with_nonce(0);
- }
+ virtual int start();
virtual void wait();
void set_cluster_protocol(int p) {
// start up network
messenger = new SimpleMessenger(g_ceph_context, entity_name_t::CLIENT());
- messenger->start_with_nonce(getpid());
+ messenger->set_nonce(getpid());
+ messenger->start();
ctx->dispatcher = new Admin(ctx.get());
messenger->add_dispatcher_head(ctx->dispatcher);