1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
16 #ifdef HAVE_SYS_MOUNT_H
17 #include <sys/mount.h>
20 #ifdef HAVE_SYS_PARAM_H
21 #include <sys/param.h>
28 #include <boost/scoped_ptr.hpp>
29 #include <boost/random/mersenne_twister.hpp>
30 #include <boost/random/uniform_int.hpp>
33 #include "osd/osd_types.h"
35 #include "osdc/Objecter.h"
36 #include "mon/MonClient.h"
37 #include "msg/Dispatcher.h"
38 #include "msg/Messenger.h"
39 #include "common/Timer.h"
40 #include "common/ceph_argparse.h"
41 #include "global/global_init.h"
42 #include "global/signal_handler.h"
43 #include "common/config.h"
44 #include "common/debug.h"
45 #include "common/errno.h"
46 #include "common/Cond.h"
47 #include "common/Mutex.h"
48 #include "common/strtol.h"
49 #include "common/LogEntry.h"
50 #include "auth/KeyRing.h"
51 #include "auth/AuthAuthorizeHandler.h"
52 #include "include/uuid.h"
53 #include "include/assert.h"
55 #include "messages/MOSDBoot.h"
56 #include "messages/MOSDAlive.h"
57 #include "messages/MOSDPGCreate.h"
58 #include "messages/MOSDPGRemove.h"
59 #include "messages/MOSDMap.h"
60 #include "messages/MPGStats.h"
61 #include "messages/MLog.h"
62 #include "messages/MOSDPGTemp.h"
66 #define dout_context g_ceph_context
67 #define dout_subsys ceph_subsys_
69 #define dout_prefix _prefix(_dout, get_name())
70 static ostream
& _prefix(std::ostream
*_dout
, string n
) {
71 return *_dout
<< " stub(" << n
<< ") ";
75 typedef boost::mt11213b rngen_t
;
76 typedef boost::scoped_ptr
<Messenger
> MessengerRef
;
77 typedef boost::scoped_ptr
<Objecter
> ObjecterRef
;
79 class TestStub
: public Dispatcher
82 MessengerRef messenger
;
92 struct C_Tick
: public Context
{
94 explicit C_Tick(TestStub
*stub
) : s(stub
) {}
95 void finish(int r
) override
{
96 generic_dout(20) << "C_Tick::" << __func__
<< dendl
;
97 if (r
== -ECANCELED
) {
98 generic_dout(20) << "C_Tick::" << __func__
99 << " shutdown" << dendl
;
106 bool ms_dispatch(Message
*m
) override
= 0;
107 void ms_handle_connect(Connection
*con
) override
= 0;
108 void ms_handle_remote_reset(Connection
*con
) override
= 0;
109 virtual int _shutdown() = 0;
110 // courtesy method to be implemented by the stubs at their
112 virtual void _tick() { }
113 // different stubs may have different needs; if a stub needs
114 // to tick, then it must call this function.
115 void start_ticking(double t
=1.0) {
121 dout(20) << __func__
<< " adding tick timer" << dendl
;
122 timer
.add_event_after(tick_seconds
, new C_Tick(this));
124 // If we have a function to start ticking that the stubs can
125 // use at their own discretion, then we should also have a
126 // function to disable said ticking to be used the same way.
128 // For simplicity's sake, we don't cancel the tick right off
129 // the bat; instead, we wait for the next tick to kick in and
131 void stop_ticking() {
132 dout(20) << __func__
<< " disable tick" << dendl
;
138 std::cout
<< __func__
<< std::endl
;
139 if (do_shutdown
|| (tick_seconds
<= 0)) {
140 std::cout
<< __func__
<< " "
141 << (do_shutdown
? "shutdown" : "stop ticking")
146 timer
.add_event_after(tick_seconds
, new C_Tick(this));
149 virtual const string
get_name() = 0;
150 virtual int init() = 0;
152 virtual int shutdown() {
153 Mutex::Locker
l(lock
);
157 dout(10) << __func__
<< " error shutting down: "
158 << cpp_strerror(-r
) << dendl
;
163 messenger
->shutdown();
167 virtual void print(ostream
&out
) {
168 out
<< "stub(" << get_name() << ")";
172 if (messenger
!= NULL
)
176 TestStub(CephContext
*cct
, string who
)
179 lock(who
.append("::lock").c_str()),
182 tick_seconds(0.0) { }
185 class ClientStub
: public TestStub
187 ObjecterRef objecter
;
191 bool ms_dispatch(Message
*m
) override
{
192 Mutex::Locker
l(lock
);
193 dout(1) << "client::" << __func__
<< " " << *m
<< dendl
;
194 switch (m
->get_type()) {
195 case CEPH_MSG_OSD_MAP
:
196 objecter
->handle_osd_map((MOSDMap
*)m
);
203 void ms_handle_connect(Connection
*con
) override
{
204 dout(1) << "client::" << __func__
<< " " << con
<< dendl
;
205 Mutex::Locker
l(lock
);
206 objecter
->ms_handle_connect(con
);
209 void ms_handle_remote_reset(Connection
*con
) override
{
210 dout(1) << "client::" << __func__
<< " " << con
<< dendl
;
211 Mutex::Locker
l(lock
);
212 objecter
->ms_handle_remote_reset(con
);
215 bool ms_handle_reset(Connection
*con
) override
{
216 dout(1) << "client::" << __func__
<< dendl
;
217 Mutex::Locker
l(lock
);
218 objecter
->ms_handle_reset(con
);
222 bool ms_handle_refused(Connection
*con
) override
{
226 const string
get_name() override
{
230 int _shutdown() override
{
232 objecter
->shutdown();
238 explicit ClientStub(CephContext
*cct
)
239 : TestStub(cct
, "client"),
240 gen((int) time(NULL
))
243 int init() override
{
245 err
= monc
.build_initial_monmap();
247 derr
<< "ClientStub::" << __func__
<< " ERROR: build initial monmap: "
248 << cpp_strerror(err
) << dendl
;
252 messenger
.reset(Messenger::create_client_messenger(cct
, "stubclient"));
253 assert(messenger
.get() != NULL
);
255 messenger
->set_default_policy(
256 Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX
));
257 dout(10) << "ClientStub::" << __func__
<< " starting messenger at "
258 << messenger
->get_myaddr() << dendl
;
260 objecter
.reset(new Objecter(cct
, messenger
.get(), &monc
, NULL
, 0, 0));
261 assert(objecter
.get() != NULL
);
262 objecter
->set_balanced_budget();
264 monc
.set_messenger(messenger
.get());
266 messenger
->add_dispatcher_head(this);
268 monc
.set_want_keys(CEPH_ENTITY_TYPE_MON
|CEPH_ENTITY_TYPE_OSD
);
272 derr
<< "ClientStub::" << __func__
<< " monc init error: "
273 << cpp_strerror(-err
) << dendl
;
277 err
= monc
.authenticate();
279 derr
<< "ClientStub::" << __func__
<< " monc authenticate error: "
280 << cpp_strerror(-err
) << dendl
;
284 monc
.wait_auth_rotating(30.0);
286 objecter
->set_client_incarnation(0);
295 objecter
->wait_for_osd_map();
297 dout(10) << "ClientStub::" << __func__
<< " done" << dendl
;
302 typedef boost::scoped_ptr
<AuthAuthorizeHandlerRegistry
> AuthHandlerRef
;
303 class OSDStub
: public TestStub
305 AuthHandlerRef auth_handler_registry
;
311 map
<pg_t
,pg_stat_t
> pgs
;
312 set
<pg_t
> pgs_changes
;
315 boost::uniform_int
<> mon_osd_rng
;
317 utime_t last_boot_attempt
;
318 static const double STUB_BOOT_INTERVAL
;
324 STUB_MON_OSD_ALIVE
= 1,
325 STUB_MON_OSD_PGTEMP
= 2,
326 STUB_MON_OSD_FAILURE
= 3,
327 STUB_MON_OSD_PGSTATS
= 4,
330 STUB_MON_OSD_FIRST
= STUB_MON_OSD_ALIVE
,
331 STUB_MON_OSD_LAST
= STUB_MON_LOG
,
334 struct C_CreatePGs
: public Context
{
336 explicit C_CreatePGs(OSDStub
*stub
) : s(stub
) {}
337 void finish(int r
) override
{
338 if (r
== -ECANCELED
) {
339 generic_dout(20) << "C_CreatePGs::" << __func__
340 << " shutdown" << dendl
;
343 generic_dout(20) << "C_CreatePGs::" << __func__
<< dendl
;
344 s
->auto_create_pgs();
349 OSDStub(int _whoami
, CephContext
*cct
)
350 : TestStub(cct
, "osd"),
351 auth_handler_registry(new AuthAuthorizeHandlerRegistry(
353 cct
->_conf
->auth_cluster_required
.length() ?
354 cct
->_conf
->auth_cluster_required
:
355 cct
->_conf
->auth_supported
)),
358 mon_osd_rng(STUB_MON_OSD_FIRST
, STUB_MON_OSD_LAST
)
360 dout(20) << __func__
<< " auth supported: "
361 << cct
->_conf
->auth_supported
<< dendl
;
363 ss
<< "client-osd" << whoami
;
364 std::string public_msgr_type
= cct
->_conf
->ms_public_type
.empty() ? cct
->_conf
->get_val
<std::string
>("ms_type") : cct
->_conf
->ms_public_type
;
365 messenger
.reset(Messenger::create(cct
, public_msgr_type
, entity_name_t::OSD(whoami
),
366 ss
.str().c_str(), getpid(), 0));
368 Throttle
throttler(g_ceph_context
, "osd_client_bytes",
369 g_conf
->osd_client_message_size_cap
);
371 messenger
->set_default_policy(
372 Messenger::Policy::stateless_server(0));
373 messenger
->set_policy_throttlers(entity_name_t::TYPE_CLIENT
,
375 messenger
->set_policy(entity_name_t::TYPE_MON
,
376 Messenger::Policy::lossy_client(
378 CEPH_FEATURE_PGID64
|
379 CEPH_FEATURE_OSDENC
));
380 messenger
->set_policy(entity_name_t::TYPE_OSD
,
381 Messenger::Policy::stateless_server(0));
383 dout(10) << __func__
<< " public addr " << g_conf
->public_addr
<< dendl
;
384 int err
= messenger
->bind(g_conf
->public_addr
);
388 if (monc
.build_initial_monmap() < 0)
392 monc
.set_messenger(messenger
.get());
395 int init() override
{
396 dout(10) << __func__
<< dendl
;
397 Mutex::Locker
l(lock
);
399 dout(1) << __func__
<< " fsid " << monc
.monmap
.fsid
400 << " osd_fsid " << g_conf
->osd_uuid
<< dendl
;
401 dout(1) << __func__
<< " name " << g_conf
->name
<< dendl
;
404 messenger
->add_dispatcher_head(this);
405 monc
.set_want_keys(CEPH_ENTITY_TYPE_MON
| CEPH_ENTITY_TYPE_OSD
);
407 int err
= monc
.init();
409 derr
<< __func__
<< " monc init error: "
410 << cpp_strerror(-err
) << dendl
;
414 err
= monc
.authenticate();
416 derr
<< __func__
<< " monc authenticate error: "
417 << cpp_strerror(-err
) << dendl
;
421 assert(!monc
.get_fsid().is_zero());
423 monc
.wait_auth_rotating(30.0);
426 dout(10) << __func__
<< " creating osd superblock" << dendl
;
427 sb
.cluster_fsid
= monc
.monmap
.fsid
;
428 sb
.osd_fsid
.generate_random();
430 sb
.compat_features
= CompatSet();
431 dout(20) << __func__
<< " " << sb
<< dendl
;
432 dout(20) << __func__
<< " osdmap " << osdmap
<< dendl
;
437 // give a chance to the mons to inform us of what PGs we should create
438 timer
.add_event_after(30.0, new C_CreatePGs(this));
443 int _shutdown() override
{
449 dout(1) << __func__
<< " boot?" << dendl
;
451 utime_t now
= ceph_clock_now();
452 if ((last_boot_attempt
> 0.0)
453 && ((now
- last_boot_attempt
)) <= STUB_BOOT_INTERVAL
) {
454 dout(1) << __func__
<< " backoff and try again later." << dendl
;
458 dout(1) << __func__
<< " boot!" << dendl
;
459 MOSDBoot
*mboot
= new MOSDBoot
;
461 last_boot_attempt
= now
;
462 monc
.send_mon_message(mboot
);
465 void add_pg(pg_t pgid
, epoch_t epoch
, pg_t parent
) {
467 utime_t now
= ceph_clock_now();
471 s
.last_epoch_clean
= epoch
;
473 s
.state
|= PG_STATE_CLEAN
| PG_STATE_ACTIVE
;
478 s
.last_unstale
= now
;
481 pgs_changes
.insert(pgid
);
484 void auto_create_pgs() {
485 bool has_pgs
= !pgs
.empty();
487 << ": " << (has_pgs
? "has pgs; ignore" : "create pgs") << dendl
;
491 if (!osdmap
.get_epoch()) {
493 << " still don't have osdmap; reschedule pg creation" << dendl
;
494 timer
.add_event_after(10.0, new C_CreatePGs(this));
498 auto& osdmap_pools
= osdmap
.get_pools();
499 for (auto pit
= osdmap_pools
.begin(); pit
!= osdmap_pools
.end(); ++pit
) {
500 const int64_t pool_id
= pit
->first
;
501 const pg_pool_t
&pool
= pit
->second
;
502 int ruleno
= pool
.get_crush_ruleset();
504 if (!osdmap
.crush
->rule_exists(ruleno
)) {
506 << " no crush rule for pool id " << pool_id
507 << " rule no " << ruleno
<< dendl
;
511 epoch_t pool_epoch
= pool
.get_last_change();
513 << " pool num pgs " << pool
.get_pg_num()
514 << " epoch " << pool_epoch
<< dendl
;
516 for (ps_t ps
= 0; ps
< pool
.get_pg_num(); ++ps
) {
517 pg_t
pgid(ps
, pool_id
, -1);
520 << " pgid " << pgid
<< " parent " << parent
<< dendl
;
521 add_pg(pgid
, pool_epoch
, parent
);
526 void update_osd_stat() {
528 int ret
= statfs(".", &stbuf
);
532 << " cannot statfs ." << cpp_strerror(ret
) << dendl
;
536 osd_stat
.kb
= stbuf
.f_blocks
* stbuf
.f_bsize
/ 1024;
537 osd_stat
.kb_used
= (stbuf
.f_blocks
- stbuf
.f_bfree
) * stbuf
.f_bsize
/ 1024;
538 osd_stat
.kb_avail
= stbuf
.f_bavail
* stbuf
.f_bsize
/ 1024;
541 void send_pg_stats() {
543 << " pgs " << pgs
.size() << " osdmap " << osdmap
<< dendl
;
544 utime_t now
= ceph_clock_now();
545 MPGStats
*mstats
= new MPGStats(monc
.get_fsid(), osdmap
.get_epoch(), now
);
548 mstats
->osd_stat
= osd_stat
;
550 set
<pg_t
>::iterator it
;
551 for (it
= pgs_changes
.begin(); it
!= pgs_changes
.end(); ++it
) {
553 if (pgs
.count(pgid
) == 0) {
555 << " pgid " << pgid
<< " not on our map" << dendl
;
556 assert(0 == "pgid not on our map");
558 pg_stat_t
&s
= pgs
[pgid
];
559 mstats
->pg_stat
[pgid
] = s
;
561 JSONFormatter
f(true);
564 << " pg " << pgid
<< " stats:\n";
569 dout(10) << __func__
<< " send " << *mstats
<< dendl
;
570 monc
.send_mon_message(mstats
);
573 void modify_pg(pg_t pgid
) {
574 dout(10) << __func__
<< " pg " << pgid
<< dendl
;
575 assert(pgs
.count(pgid
) > 0);
577 pg_stat_t
&s
= pgs
[pgid
];
578 utime_t now
= ceph_clock_now();
580 if (now
- s
.last_change
< 10.0) {
582 << " pg " << pgid
<< " changed in the last 10s" << dendl
;
586 s
.state
^= PG_STATE_CLEAN
;
587 if (s
.state
& PG_STATE_CLEAN
)
592 pgs_changes
.insert(pgid
);
596 dout(10) << __func__
<< dendl
;
600 << " no pgs available! don't attempt to modify." << dendl
;
604 boost::uniform_int
<> pg_rng(0, pgs
.size()-1);
607 int num_pgs
= pg_rng(gen
);
608 while ((int)pgs_pos
.size() < num_pgs
)
609 pgs_pos
.insert(pg_rng(gen
));
611 map
<pg_t
,pg_stat_t
>::iterator it
= pgs
.begin();
612 set
<int>::iterator pos_it
= pgs_pos
.begin();
615 while (pos_it
!= pgs_pos
.end()) {
617 dout(20) << __func__
<< " pg at pos " << at
<< dendl
;
618 while ((pgs_at
!= at
) && (it
!= pgs
.end())) {
622 assert(it
!= pgs
.end());
624 << " pg at pos " << at
<< ": " << it
->first
<< dendl
;
625 modify_pg(it
->first
);
631 dout(10) << __func__
<< dendl
;
632 if (!osdmap
.exists(whoami
)) {
633 dout(0) << __func__
<< " I'm not in the osdmap!!\n";
634 JSONFormatter
f(true);
639 if (osdmap
.get_epoch() == 0) {
640 dout(1) << __func__
<< " wait for osdmap" << dendl
;
643 epoch_t up_thru
= osdmap
.get_up_thru(whoami
);
644 dout(10) << __func__
<< "up_thru: " << osdmap
.get_up_thru(whoami
) << dendl
;
646 monc
.send_mon_message(new MOSDAlive(osdmap
.get_epoch(), up_thru
));
650 if (osdmap
.get_epoch() == 0) {
651 dout(1) << __func__
<< " wait for osdmap" << dendl
;
654 dout(10) << __func__
<< dendl
;
655 MOSDPGTemp
*m
= new MOSDPGTemp(osdmap
.get_epoch());
656 monc
.send_mon_message(m
);
660 dout(10) << __func__
<< dendl
;
664 dout(10) << __func__
<< dendl
;
667 if (!pgs_changes
.empty())
669 monc
.sub_want("osd_pg_creates", 0, CEPH_SUBSCRIBE_ONETIME
);
672 dout(20) << __func__
<< " pg pools:\n";
674 JSONFormatter
f(true);
675 f
.open_array_section("pools");
676 auto& osdmap_pools
= osdmap
.get_pools();
677 for (auto pit
= osdmap_pools
.begin(); pit
!= osdmap_pools
.end(); ++pit
) {
678 const int64_t pool_id
= pit
->first
;
679 const pg_pool_t
&pool
= pit
->second
;
680 f
.open_object_section("pool");
681 f
.dump_int("pool_id", pool_id
);
682 f
.open_object_section("pool_dump");
693 dout(10) << __func__
<< dendl
;
695 MLog
*m
= new MLog(monc
.get_fsid());
697 boost::uniform_int
<> log_rng(1, 10);
698 size_t num_entries
= log_rng(gen
);
700 << " send " << num_entries
<< " log messages" << dendl
;
702 utime_t now
= ceph_clock_now();
704 for (; num_entries
> 0; --num_entries
) {
706 e
.who
= messenger
->get_myinst();
710 e
.msg
= "OSDStub::op_log";
711 m
->entries
.push_back(e
);
714 monc
.send_mon_message(m
);
717 void _tick() override
{
718 if (!osdmap
.exists(whoami
)) {
719 std::cout
<< __func__
<< " not in the cluster; boot!" << std::endl
;
726 boost::uniform_int
<> op_rng(STUB_MON_OSD_FIRST
, STUB_MON_OSD_LAST
);
727 int op
= op_rng(gen
);
729 case STUB_MON_OSD_ALIVE
:
732 case STUB_MON_OSD_PGTEMP
:
735 case STUB_MON_OSD_FAILURE
:
738 case STUB_MON_OSD_PGSTATS
:
747 void handle_pg_create(MOSDPGCreate
*m
) {
749 if (m
->epoch
< osdmap
.get_epoch()) {
750 std::cout
<< __func__
<< " epoch " << m
->epoch
<< " < "
751 << osdmap
.get_epoch() << "; dropping" << std::endl
;
756 for (map
<pg_t
,pg_create_t
>::iterator it
= m
->mkpg
.begin();
757 it
!= m
->mkpg
.end(); ++it
) {
758 pg_create_t
&c
= it
->second
;
759 std::cout
<< __func__
<< " pg " << it
->first
760 << " created " << c
.created
761 << " parent " << c
.parent
<< std::endl
;
762 if (pgs
.count(it
->first
)) {
763 std::cout
<< __func__
<< " pg " << it
->first
764 << " exists; skipping" << std::endl
;
768 pg_t pgid
= it
->first
;
769 add_pg(pgid
, c
.created
, c
.parent
);
774 void handle_osd_map(MOSDMap
*m
) {
775 dout(1) << __func__
<< dendl
;
776 if (m
->fsid
!= monc
.get_fsid()) {
778 << " message fsid " << m
->fsid
<< " != " << monc
.get_fsid()
780 dout(0) << __func__
<< " " << m
781 << " from " << m
->get_source_inst()
783 dout(0) << monc
.get_monmap() << dendl
;
785 assert(m
->fsid
== monc
.get_fsid());
787 epoch_t first
= m
->get_first();
788 epoch_t last
= m
->get_last();
790 << " epochs [" << first
<< "," << last
<< "]"
791 << " current " << osdmap
.get_epoch() << dendl
;
793 if (last
<= osdmap
.get_epoch()) {
794 dout(5) << __func__
<< " no new maps here; dropping" << dendl
;
799 if (first
> osdmap
.get_epoch() + 1) {
801 << osdmap
.get_epoch() + 1 << ".." << (first
-1) << dendl
;
802 if ((m
->oldest_map
< first
&& osdmap
.get_epoch() == 0) ||
803 m
->oldest_map
<= osdmap
.get_epoch()) {
804 monc
.sub_want("osdmap", osdmap
.get_epoch()+1,
805 CEPH_SUBSCRIBE_ONETIME
);
812 epoch_t start_full
= MAX(osdmap
.get_epoch() + 1, first
);
814 if (m
->maps
.size() > 0) {
815 map
<epoch_t
,bufferlist
>::reverse_iterator rit
;
816 rit
= m
->maps
.rbegin();
817 if (start_full
<= rit
->first
) {
818 start_full
= rit
->first
;
820 << " full epoch " << start_full
<< dendl
;
821 bufferlist
&bl
= rit
->second
;
822 bufferlist::iterator p
= bl
.begin();
827 for (epoch_t e
= start_full
; e
<= last
; e
++) {
828 map
<epoch_t
,bufferlist
>::iterator it
;
829 it
= m
->incremental_maps
.find(e
);
830 if (it
== m
->incremental_maps
.end())
834 << " incremental epoch " << e
835 << " on full epoch " << start_full
<< dendl
;
836 OSDMap::Incremental inc
;
837 bufferlist
&bl
= it
->second
;
838 bufferlist::iterator p
= bl
.begin();
841 int err
= osdmap
.apply_incremental(inc
);
843 derr
<< "osd." << whoami
<< "::" << __func__
844 << "** ERROR: applying incremental: "
845 << cpp_strerror(err
) << dendl
;
846 assert(0 == "error applying incremental");
849 dout(30) << __func__
<< "\nosdmap:\n";
850 JSONFormatter
f(true);
855 if (osdmap
.is_up(whoami
) &&
856 osdmap
.get_addr(whoami
) == messenger
->get_myaddr()) {
858 << " got into the osdmap and we're up!" << dendl
;
861 if (m
->newest_map
&& m
->newest_map
> last
) {
863 << " they have more maps; requesting them!" << dendl
;
864 monc
.sub_want("osdmap", osdmap
.get_epoch()+1, CEPH_SUBSCRIBE_ONETIME
);
868 dout(10) << __func__
<< " done" << dendl
;
872 bool ms_dispatch(Message
*m
) override
{
873 dout(1) << __func__
<< " " << *m
<< dendl
;
875 switch (m
->get_type()) {
876 case MSG_OSD_PG_CREATE
:
877 handle_pg_create((MOSDPGCreate
*)m
);
879 case CEPH_MSG_OSD_MAP
:
880 handle_osd_map((MOSDMap
*)m
);
889 void ms_handle_connect(Connection
*con
) override
{
890 dout(1) << __func__
<< " " << con
<< dendl
;
891 if (con
->get_peer_type() == CEPH_ENTITY_TYPE_MON
) {
892 dout(10) << __func__
<< " on mon" << dendl
;
896 void ms_handle_remote_reset(Connection
*con
) override
{}
898 bool ms_handle_reset(Connection
*con
) override
{
899 dout(1) << __func__
<< dendl
;
900 Session
*session
= (Session
*)con
->get_priv();
907 bool ms_handle_refused(Connection
*con
) override
{
911 const string
get_name() override
{
913 ss
<< "osd." << whoami
;
918 double const OSDStub::STUB_BOOT_INTERVAL
= 10.0;
921 #define dout_prefix *_dout << "main "
923 const char *our_name
= NULL
;
924 vector
<TestStub
*> stubs
;
925 Mutex
shutdown_lock("main::shutdown_lock");
927 Context
*shutdown_cb
= NULL
;
928 SafeTimer
*shutdown_timer
= NULL
;
930 struct C_Shutdown
: public Context
932 void finish(int r
) override
{
933 generic_dout(10) << "main::shutdown time has ran out" << dendl
;
934 shutdown_cond
.Signal();
938 void handle_test_signal(int signum
)
940 if ((signum
!= SIGINT
) && (signum
!= SIGTERM
))
943 std::cerr
<< "*** Got signal " << sig_str(signum
) << " ***" << std::endl
;
944 Mutex::Locker
l(shutdown_lock
);
945 if (shutdown_timer
) {
946 shutdown_timer
->cancel_all_events();
947 shutdown_cond
.Signal();
952 assert(our_name
!= NULL
);
954 std::cout
<< "usage: " << our_name
955 << " <--stub-id ID> [--stub-id ID...]"
959 -c FILE Read configuration from FILE\n\
960 --keyring FILE Read keyring from FILE\n\
961 --help This message\n\
963 Test-specific Options:\n\
964 --stub-id ID1..ID2 Interval of OSD ids for multiple stubs to mimic.\n\
965 --stub-id ID OSD id a stub will mimic to be\n\
966 (same as --stub-id ID..ID)\n\
970 int get_id_interval(int &first
, int &last
, string
&str
)
972 size_t found
= str
.find("..");
973 string first_str
, last_str
;
974 if (found
== string::npos
) {
975 first_str
= last_str
= str
;
977 first_str
= str
.substr(0, found
);
978 last_str
= str
.substr(found
+2);
982 first
= strict_strtol(first_str
.c_str(), 10, &err
);
983 if ((first
== 0) && (!err
.empty())) {
984 std::cerr
<< err
<< std::endl
;
988 last
= strict_strtol(last_str
.c_str(), 10, &err
);
989 if ((last
== 0) && (!err
.empty())) {
990 std::cerr
<< err
<< std::endl
;
996 int main(int argc
, const char *argv
[])
998 vector
<const char*> def_args
;
999 vector
<const char*> args
;
1001 argv_to_vec(argc
, argv
, args
);
1003 auto cct
= global_init(&def_args
, args
,
1004 CEPH_ENTITY_TYPE_OSD
, CODE_ENVIRONMENT_UTILITY
,
1007 common_init_finish(g_ceph_context
);
1008 g_ceph_context
->_conf
->apply_changes(NULL
);
1011 double duration
= 300.0;
1013 for (std::vector
<const char*>::iterator i
= args
.begin(); i
!= args
.end();) {
1016 if (ceph_argparse_double_dash(args
, i
)) {
1018 } else if (ceph_argparse_witharg(args
, i
, &val
,
1019 "--stub-id", (char*) NULL
)) {
1020 int first
= -1, last
= -1;
1021 if (get_id_interval(first
, last
, val
) < 0) {
1022 std::cerr
<< "** error parsing stub id '" << val
<< "'" << std::endl
;
1026 for (; first
<= last
; ++first
)
1027 stub_ids
.insert(first
);
1028 } else if (ceph_argparse_witharg(args
, i
, &val
,
1029 "--duration", (char*) NULL
)) {
1031 duration
= (double) strict_strtol(val
.c_str(), 10, &err
);
1032 if ((duration
== 0) && (!err
.empty())) {
1033 std::cerr
<< "** error parsing '--duration " << val
<< "': '"
1034 << err
<< std::endl
;
1037 } else if (ceph_argparse_flag(args
, i
, "--help", (char*) NULL
)) {
1041 std::cerr
<< "unknown argument '" << *i
<< "'" << std::endl
;
1046 if (stub_ids
.empty()) {
1047 std::cerr
<< "** error: must specify at least one '--stub-id <ID>'"
1053 for (set
<int>::iterator i
= stub_ids
.begin(); i
!= stub_ids
.end(); ++i
) {
1056 std::cout
<< __func__
<< " starting stub." << whoami
<< std::endl
;
1057 OSDStub
*stub
= new OSDStub(whoami
, g_ceph_context
);
1058 int err
= stub
->init();
1060 std::cerr
<< "** osd stub error: " << cpp_strerror(-err
) << std::endl
;
1063 stubs
.push_back(stub
);
1066 std::cout
<< __func__
<< " starting client stub" << std::endl
;
1067 ClientStub
*cstub
= new ClientStub(g_ceph_context
);
1068 int err
= cstub
->init();
1070 std::cerr
<< "** client stub error: " << cpp_strerror(-err
) << std::endl
;
1073 stubs
.push_back(cstub
);
1075 init_async_signal_handler();
1076 register_async_signal_handler_oneshot(SIGINT
, handle_test_signal
);
1077 register_async_signal_handler_oneshot(SIGTERM
, handle_test_signal
);
1079 shutdown_lock
.Lock();
1080 shutdown_timer
= new SafeTimer(g_ceph_context
, shutdown_lock
);
1081 shutdown_timer
->init();
1082 if (duration
!= 0) {
1083 std::cout
<< __func__
1084 << " run test for " << duration
<< " seconds" << std::endl
;
1085 shutdown_timer
->add_event_after((double) duration
, new C_Shutdown
);
1087 shutdown_cond
.Wait(shutdown_lock
);
1089 shutdown_timer
->shutdown();
1090 delete shutdown_timer
;
1091 shutdown_timer
= NULL
;
1092 shutdown_lock
.Unlock();
1094 unregister_async_signal_handler(SIGINT
, handle_test_signal
);
1095 unregister_async_signal_handler(SIGTERM
, handle_test_signal
);
1097 std::cout
<< __func__
<< " waiting for stubs to finish" << std::endl
;
1098 vector
<TestStub
*>::iterator it
;
1100 for (i
= 0, it
= stubs
.begin(); it
!= stubs
.end(); ++it
, ++i
) {
1104 std::cout
<< __func__
<< " finished " << (*it
)->get_name() << std::endl
;