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.
17 #ifndef CEPH_MESSENGER_H
18 #define CEPH_MESSENGER_H
28 #include "Dispatcher.h"
30 #include "common/Throttle.h"
31 #include "include/Context.h"
32 #include "include/types.h"
33 #include "include/ceph_features.h"
34 #include "auth/Crypto.h"
35 #include "common/item_history.h"
36 #include "auth/AuthRegistry.h"
37 #include "include/ceph_assert.h"
43 #define SOCKET_PRIORITY_MIN_DELAY 6
50 #ifdef UNIT_TESTS_BUILT
54 std::condition_variable cond_var
;
56 enum ACTION
: uint32_t {
62 virtual ~Interceptor() {}
63 virtual ACTION
intercept(Connection
*conn
, uint32_t step
) = 0;
70 std::deque
<Dispatcher
*> dispatchers
;
71 std::deque
<Dispatcher
*> fast_dispatchers
;
72 ZTracer::Endpoint trace_endpoint
;
75 void set_endpoint_addr(const entity_addr_t
& a
,
76 const entity_name_t
&name
);
79 /// the "name" of the local daemon. eg client.99
80 entity_name_t my_name
;
83 safe_item_history
<entity_addrvec_t
> my_addrs
;
85 int default_send_priority
;
86 /// std::set to true once the Messenger has started, and std::set to false on shutdown
92 AuthClient
*auth_client
= 0;
93 AuthServer
*auth_server
= 0;
95 #ifdef UNIT_TESTS_BUILT
96 Interceptor
*interceptor
= nullptr;
100 * Various Messenger conditional config/type flags to allow
101 * different "transport" Messengers to tune themselves
103 static const int HAS_HEAVY_TRAFFIC
= 0x0001;
104 static const int HAS_MANY_CONNECTIONS
= 0x0002;
105 static const int HEARTBEAT
= 0x0004;
108 * The CephContext this Messenger uses. Many other components initialize themselves
114 using Policy
= ceph::net::Policy
<Throttle
>;
117 // allow unauthenticated connections. This is needed for
118 // compatibility with pre-nautilus OSDs, which do not authenticate
119 // the heartbeat sessions.
120 bool require_authorizer
= true;
123 // for authentication
124 AuthRegistry auth_registry
;
128 * Messenger constructor. Call this from your implementation.
129 * Messenger users should construct full implementations directly,
130 * or use the create() function.
132 Messenger(CephContext
*cct_
, entity_name_t w
);
133 virtual ~Messenger() {}
136 * create a new messenger
138 * Create a new messenger instance, with whatever implementation is
139 * available or specified via the configuration in cct.
142 * @param type name of messenger type
143 * @param name entity name to register
144 * @param lname logical name of the messenger in this process (e.g., "client")
145 * @param nonce nonce value to uniquely identify this instance on the current host
146 * @param features bits for the local connection
147 * @param cflags general std::set of flags to configure transport resources
149 static Messenger
*create(CephContext
*cct
,
150 const std::string
&type
,
156 static uint64_t get_random_nonce();
157 static uint64_t get_pid_nonce();
160 * create a new messenger
162 * Create a new messenger instance.
163 * Same as the above, but a slightly simpler interface for clients:
164 * - Generate a random nonce
165 * - use the default feature bits
166 * - get the messenger type from cct
167 * - use the client entity_type
170 * @param lname logical name of the messenger in this process (e.g., "client")
172 static Messenger
*create_client_messenger(CephContext
*cct
, std::string lname
);
175 * @defgroup Accessors
178 int get_mytype() const { return my_name
.type(); }
181 * Retrieve the Messenger's name
183 * @return A const reference to the name this Messenger
184 * currently believes to be its own.
186 const entity_name_t
& get_myname() { return my_name
; }
189 * Retrieve the Messenger's address.
191 * @return A const reference to the address this Messenger
192 * currently believes to be its own.
194 const entity_addrvec_t
& get_myaddrs() {
199 * get legacy addr for myself, suitable for protocol v1
201 * Note that myaddrs might be a proper addrvec with v1 in it, or it might be an
202 * ANY addr (if i am a pure client).
204 entity_addr_t
get_myaddr_legacy() {
205 return my_addrs
->as_legacy_addr();
210 * std::set messenger's instance
212 uint32_t get_magic() { return magic
; }
213 void set_magic(int _magic
) { magic
= _magic
; }
215 void set_auth_client(AuthClient
*ac
) {
218 void set_auth_server(AuthServer
*as
) {
224 * std::set messenger's address
226 virtual void set_myaddrs(const entity_addrvec_t
& a
) {
228 set_endpoint_addr(a
.front(), my_name
);
232 * @return the zipkin trace endpoint
234 const ZTracer::Endpoint
* get_trace_endpoint() const {
235 return &trace_endpoint
;
239 * set the name of the local entity. The name is reported to others and
240 * can be changed while the system is running, but doing so at incorrect
241 * times may have bad results.
243 * @param m The name to std::set.
245 void set_myname(const entity_name_t
& m
) { my_name
= m
; }
248 * set the unknown address components for this Messenger.
249 * This is useful if the Messenger doesn't know its full address just by
250 * binding, but another Messenger on the same interface has already learned
251 * its full address. This function does not fill in known address elements,
252 * cause a rebind, or do anything of that sort.
254 * @param addr The address to use as a template.
256 virtual bool set_addr_unknowns(const entity_addrvec_t
&addrs
) = 0;
258 * set the address for this Messenger. This is useful if the Messenger
259 * binds to a specific address but advertises a different address on the
262 * @param addr The address to use.
264 virtual void set_addrs(const entity_addrvec_t
&addr
) = 0;
265 /// Get the default send priority.
266 int get_default_send_priority() { return default_send_priority
; }
268 * Get the number of Messages which the Messenger has received
269 * but not yet dispatched.
271 virtual int get_dispatch_queue_len() = 0;
274 * Get age of oldest undelivered message
275 * (0 if the queue is empty)
277 virtual double get_dispatch_queue_max_age(utime_t now
) = 0;
284 * @defgroup Configuration
288 * set the cluster protocol in use by this daemon.
289 * This is an init-time function and cannot be called after calling
292 * @param p The cluster protocol to use. Defined externally.
294 virtual void set_cluster_protocol(int p
) = 0;
296 * set a policy which is applied to all peers who do not have a type-specific
298 * This is an init-time function and cannot be called after calling
301 * @param p The Policy to apply.
303 virtual void set_default_policy(Policy p
) = 0;
305 * set a policy which is applied to all peers of the given type.
306 * This is an init-time function and cannot be called after calling
309 * @param type The peer type this policy applies to.
310 * @param p The policy to apply.
312 virtual void set_policy(int type
, Policy p
) = 0;
314 * set the Policy associated with a type of peer.
316 * This can be called either on initial setup, or after connections
317 * are already established. However, the policies for existing
318 * connections will not be affected; the new policy will only apply
319 * to future connections.
321 * @param t The peer type to get the default policy for.
322 * @return A const Policy reference.
324 virtual Policy
get_policy(int t
) = 0;
326 * Get the default Policy
328 * @return A const Policy reference.
330 virtual Policy
get_default_policy() = 0;
332 * set Throttlers applied to all Messages from the given type of peer
334 * This is an init-time function and cannot be called after calling
337 * @param type The peer type the Throttlers will apply to.
338 * @param bytes The Throttle for the number of bytes carried by the message
339 * @param msgs The Throttle for the number of messages for this @p type
340 * @note The Messenger does not take ownership of the Throttle pointers, but
341 * you must not destroy them before you destroy the Messenger.
343 virtual void set_policy_throttlers(int type
, Throttle
*bytes
, Throttle
*msgs
=NULL
) = 0;
345 * set the default send priority
347 * This is an init-time function and must be called *before* calling
350 * @param p The cluster protocol to use. Defined externally.
352 void set_default_send_priority(int p
) {
353 ceph_assert(!started
);
354 default_send_priority
= p
;
357 * set the priority(SO_PRIORITY) for all packets to be sent on this socket.
359 * Linux uses this value to order the networking queues: packets with a higher
360 * priority may be processed first depending on the selected device queueing
363 * @param prio The priority. Setting a priority outside the range 0 to 6
364 * requires the CAP_NET_ADMIN capability.
366 void set_socket_priority(int prio
) {
367 socket_priority
= prio
;
370 * Get the socket priority
372 * @return the socket priority
374 int get_socket_priority() {
375 return socket_priority
;
378 * Add a new Dispatcher to the front of the list. If you add
379 * a Dispatcher which is already included, it will get a duplicate
380 * entry. This will reduce efficiency but not break anything.
382 * @param d The Dispatcher to insert into the list.
384 void add_dispatcher_head(Dispatcher
*d
) {
385 bool first
= dispatchers
.empty();
386 dispatchers
.push_front(d
);
387 if (d
->ms_can_fast_dispatch_any())
388 fast_dispatchers
.push_front(d
);
393 * Add a new Dispatcher to the end of the list. If you add
394 * a Dispatcher which is already included, it will get a duplicate
395 * entry. This will reduce efficiency but not break anything.
397 * @param d The Dispatcher to insert into the list.
399 void add_dispatcher_tail(Dispatcher
*d
) {
400 bool first
= dispatchers
.empty();
401 dispatchers
.push_back(d
);
402 if (d
->ms_can_fast_dispatch_any())
403 fast_dispatchers
.push_back(d
);
408 * Bind the Messenger to a specific address. If bind_addr
409 * is not completely filled in the system will use the
410 * valid portions and cycle through the unset ones (eg, the port)
411 * in an unspecified order.
413 * @param bind_addr The address to bind to.
414 * @return 0 on success, or -1 on error, or -errno if
415 * we can be more specific about the failure.
417 virtual int bind(const entity_addr_t
& bind_addr
) = 0;
420 * This function performs a full restart of the Messenger component,
421 * whatever that means. Other entities who connect to this
422 * Messenger post-rebind() should perceive it as a new entity which
423 * they have not previously contacted, and it MUST bind to a
424 * different address than it did previously.
426 * @param avoid_ports Additional port to avoid binding to.
428 virtual int rebind(const std::set
<int>& avoid_ports
) { return -EOPNOTSUPP
; }
430 * Bind the 'client' Messenger to a specific address.Messenger will bind
431 * the address before connect to others when option ms_bind_before_connect
433 * @param bind_addr The address to bind to.
434 * @return 0 on success, or -1 on error, or -errno if
436 virtual int client_bind(const entity_addr_t
& bind_addr
) = 0;
438 virtual int bindv(const entity_addrvec_t
& addrs
);
441 virtual bool should_use_msgr2() {
446 * @} // Configuration
450 * @defgroup Startup/Shutdown
454 * Perform any resource allocation, thread startup, etc
455 * that is required before attempting to connect to other
456 * Messengers or transmit messages.
457 * Once this function completes, started shall be set to true.
459 * @return 0 on success; -errno on failure.
461 virtual int start() { started
= true; return 0; }
465 * Block until the Messenger has finished shutting down (according
466 * to the shutdown() function).
467 * It is valid to call this after calling shutdown(), but it must
468 * be called before deleting the Messenger.
470 virtual void wait() = 0;
472 * Initiate a shutdown of the Messenger.
474 * @return 0 on success, -errno otherwise.
476 virtual int shutdown() { started
= false; return 0; }
478 * @} // Startup/Shutdown
482 * @defgroup Messaging
486 * Queue the given Message for the given entity.
487 * Success in this function does not guarantee Message delivery, only
488 * success in queueing the Message. Other guarantees may be provided based
489 * on the Connection policy associated with the dest.
491 * @param m The Message to send. The Messenger consumes a single reference
492 * when you pass it in.
493 * @param dest The entity to send the Message to.
495 * DEPRECATED: please do not use this interface for any new code;
496 * use the Connection* variant.
498 * @return 0 on success, or -errno on failure.
503 const entity_addrvec_t
& addr
) = 0;
505 Message
*m
, const entity_addrvec_t
& addrs
) {
506 return send_to(m
, CEPH_ENTITY_TYPE_MON
, addrs
);
509 Message
*m
, const entity_addrvec_t
& addrs
) {
510 return send_to(m
, CEPH_ENTITY_TYPE_MDS
, addrs
);
517 * @defgroup Connection Management
521 * Get the Connection object associated with a given entity. If a
522 * Connection does not exist, create one and establish a logical connection.
523 * The caller owns a reference when this returns. Call ->put() when you're
526 * @param dest The entity to get a connection for.
528 virtual ConnectionRef
connect_to(
529 int type
, const entity_addrvec_t
& dest
,
530 bool anon
=false, bool not_local_dest
=false) = 0;
531 ConnectionRef
connect_to_mon(const entity_addrvec_t
& dest
,
532 bool anon
=false, bool not_local_dest
=false) {
533 return connect_to(CEPH_ENTITY_TYPE_MON
, dest
, anon
, not_local_dest
);
535 ConnectionRef
connect_to_mds(const entity_addrvec_t
& dest
,
536 bool anon
=false, bool not_local_dest
=false) {
537 return connect_to(CEPH_ENTITY_TYPE_MDS
, dest
, anon
, not_local_dest
);
539 ConnectionRef
connect_to_osd(const entity_addrvec_t
& dest
,
540 bool anon
=false, bool not_local_dest
=false) {
541 return connect_to(CEPH_ENTITY_TYPE_OSD
, dest
, anon
, not_local_dest
);
543 ConnectionRef
connect_to_mgr(const entity_addrvec_t
& dest
,
544 bool anon
=false, bool not_local_dest
=false) {
545 return connect_to(CEPH_ENTITY_TYPE_MGR
, dest
, anon
, not_local_dest
);
549 * Get the Connection object associated with ourselves.
551 virtual ConnectionRef
get_loopback_connection() = 0;
553 * Mark down a Connection to a remote.
555 * This will cause us to discard our outgoing queue for them, and if
556 * reset detection is enabled in the policy and the endpoint tries
557 * to reconnect they will discard their queue when we inform them of
560 * If there is no Connection to the given dest, it is a no-op.
562 * This generates a RESET notification to the Dispatcher.
564 * DEPRECATED: please do not use this interface for any new code;
565 * use the Connection* variant.
567 * @param a The address to mark down.
569 virtual void mark_down(const entity_addr_t
& a
) = 0;
570 virtual void mark_down_addrs(const entity_addrvec_t
& a
) {
571 mark_down(a
.legacy_addr());
574 * Mark all the existing Connections down. This is equivalent
575 * to iterating over all Connections and calling mark_down()
578 * This will generate a RESET event for each closed connections.
580 virtual void mark_down_all() = 0;
582 * @} // Connection Management
586 * @defgroup Subclass Interfacing
590 * A courtesy function for Messenger implementations which
591 * will be called when we receive our first Dispatcher.
593 virtual void ready() { }
595 * @} // Subclass Interfacing
598 #ifdef CEPH_USE_SIGPIPE_BLOCKER
600 * We need to disable SIGPIPE on all platforms, and if they
601 * don't give us a better mechanism (read: are on Solaris) that
602 * means blocking the signal whenever we do a send or sendmsg...
603 * That means any implementations must invoke MSGR_SIGPIPE_STOPPER in-scope
604 * whenever doing so. On most systems that's blank, but on systems where
605 * it's needed we construct an RAII object to plug and un-plug the SIGPIPE.
606 * See http://www.microhowto.info/howto/ignore_sigpipe_without_affecting_other_threads_in_a_process.html
608 struct sigpipe_stopper
{
610 sigset_t existing_mask
;
613 sigemptyset(&pipe_mask
);
614 sigaddset(&pipe_mask
, SIGPIPE
);
616 sigemptyset(&signals
);
617 sigpending(&signals
);
618 if (sigismember(&signals
, SIGPIPE
)) {
622 int r
= pthread_sigmask(SIG_BLOCK
, &pipe_mask
, &existing_mask
);
628 struct timespec nowait
{0};
629 int r
= sigtimedwait(&pipe_mask
, 0, &nowait
);
630 ceph_assert(r
== EAGAIN
|| r
== 0);
631 r
= pthread_sigmask(SIG_SETMASK
, &existing_mask
, 0);
636 # define MSGR_SIGPIPE_STOPPER Messenger::sigpipe_stopper stopper();
638 # define MSGR_SIGPIPE_STOPPER
641 * @defgroup Dispatcher Interfacing
645 * Determine whether a message can be fast-dispatched. We will
646 * query each Dispatcher in sequence to determine if they are
647 * capable of handling a particular message via "fast dispatch".
649 * @param m The Message we are testing.
651 bool ms_can_fast_dispatch(const cref_t
<Message
>& m
) {
652 for (const auto &dispatcher
: fast_dispatchers
) {
653 if (dispatcher
->ms_can_fast_dispatch2(m
))
660 * Deliver a single Message via "fast dispatch".
662 * @param m The Message we are fast dispatching.
663 * If none of our Dispatchers can handle it, ceph_abort().
665 void ms_fast_dispatch(const ref_t
<Message
> &m
) {
666 m
->set_dispatch_stamp(ceph_clock_now());
667 for (const auto &dispatcher
: fast_dispatchers
) {
668 if (dispatcher
->ms_can_fast_dispatch2(m
)) {
669 dispatcher
->ms_fast_dispatch2(m
);
675 void ms_fast_dispatch(Message
*m
) {
676 return ms_fast_dispatch(ref_t
<Message
>(m
, false)); /* consume ref */
681 void ms_fast_preprocess(const ref_t
<Message
> &m
) {
682 for (const auto &dispatcher
: fast_dispatchers
) {
683 dispatcher
->ms_fast_preprocess2(m
);
687 * Deliver a single Message. Send it to each Dispatcher
688 * in sequence until one of them handles it.
689 * If none of our Dispatchers can handle it, ceph_abort().
691 * @param m The Message to deliver.
693 void ms_deliver_dispatch(const ref_t
<Message
> &m
) {
694 m
->set_dispatch_stamp(ceph_clock_now());
695 for (const auto &dispatcher
: dispatchers
) {
696 if (dispatcher
->ms_dispatch2(m
))
699 lsubdout(cct
, ms
, 0) << "ms_deliver_dispatch: unhandled message " << m
<< " " << *m
<< " from "
700 << m
->get_source_inst() << dendl
;
701 ceph_assert(!cct
->_conf
->ms_die_on_unhandled_msg
);
703 void ms_deliver_dispatch(Message
*m
) {
704 return ms_deliver_dispatch(ref_t
<Message
>(m
, false)); /* consume ref */
707 * Notify each Dispatcher of a new Connection. Call
708 * this function whenever a new Connection is initiated or
711 * @param con Pointer to the new Connection.
713 void ms_deliver_handle_connect(Connection
*con
) {
714 for (const auto& dispatcher
: dispatchers
) {
715 dispatcher
->ms_handle_connect(con
);
720 * Notify each fast Dispatcher of a new Connection. Call
721 * this function whenever a new Connection is initiated or
724 * @param con Pointer to the new Connection.
726 void ms_deliver_handle_fast_connect(Connection
*con
) {
727 for (const auto& dispatcher
: fast_dispatchers
) {
728 dispatcher
->ms_handle_fast_connect(con
);
733 * Notify each Dispatcher of a new incoming Connection. Call
734 * this function whenever a new Connection is accepted.
736 * @param con Pointer to the new Connection.
738 void ms_deliver_handle_accept(Connection
*con
) {
739 for (const auto& dispatcher
: dispatchers
) {
740 dispatcher
->ms_handle_accept(con
);
745 * Notify each fast Dispatcher of a new incoming Connection. Call
746 * this function whenever a new Connection is accepted.
748 * @param con Pointer to the new Connection.
750 void ms_deliver_handle_fast_accept(Connection
*con
) {
751 for (const auto& dispatcher
: fast_dispatchers
) {
752 dispatcher
->ms_handle_fast_accept(con
);
757 * Notify each Dispatcher of a Connection which may have lost
758 * Messages. Call this function whenever you detect that a lossy Connection
759 * has been disconnected.
761 * @param con Pointer to the broken Connection.
763 void ms_deliver_handle_reset(Connection
*con
) {
764 for (const auto& dispatcher
: dispatchers
) {
765 if (dispatcher
->ms_handle_reset(con
))
770 * Notify each Dispatcher of a Connection which has been "forgotten" about
771 * by the remote end, implying that messages have probably been lost.
772 * Call this function whenever you detect a reset.
774 * @param con Pointer to the broken Connection.
776 void ms_deliver_handle_remote_reset(Connection
*con
) {
777 for (const auto& dispatcher
: dispatchers
) {
778 dispatcher
->ms_handle_remote_reset(con
);
783 * Notify each Dispatcher of a Connection for which reconnection
784 * attempts are being refused. Call this function whenever you
785 * detect that a lossy Connection has been disconnected and it's
786 * impossible to reconnect.
788 * @param con Pointer to the broken Connection.
790 void ms_deliver_handle_refused(Connection
*con
) {
791 for (const auto& dispatcher
: dispatchers
) {
792 if (dispatcher
->ms_handle_refused(con
))
797 void set_require_authorizer(bool b
) {
798 require_authorizer
= b
;
802 * @} // Dispatcher Interfacing