TransporterRegistry.hpp

/*
   Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

//****************************************************************************
//
//  NAME
//      TransporterRegistry
//
//  DESCRIPTION
//      TransporterRegistry (singelton) is the interface to the 
//      transporter layer. It handles transporter states and 
//      holds the transporter arrays.
//
//***************************************************************************/
#ifndef TransporterRegistry_H
#define TransporterRegistry_H

#if defined(HAVE_EPOLL_CREATE)
#include <sys/epoll.h>
#endif
#include "TransporterDefinitions.hpp"
#include "TransporterCallback.hpp"
#include <SocketServer.hpp>
#include <SocketClient.hpp>

#include <NdbTCP.h>

#include <mgmapi/mgmapi.h>

#include <NodeBitmask.hpp>

// A transporter is always in an IOState.
// NoHalt is used initially and as long as it is no restrictions on
// sending or receiving.
enum IOState {
  NoHalt     = 0,
  HaltInput  = 1,
  HaltOutput = 2,
  HaltIO     = 3
};


static const char *performStateString[] = 
  { "is connected",
    "is trying to connect",
    "does nothing",
    "is trying to disconnect" };

class Transporter;
class TCP_Transporter;
class SCI_Transporter;
class SHM_Transporter;

class TransporterRegistry;
class SocketAuthenticator;

class TransporterService : public SocketServer::Service {
  SocketAuthenticator * m_auth;
  TransporterRegistry * m_transporter_registry;
public:
  TransporterService(SocketAuthenticator *auth= 0)
  {
    m_auth= auth;
    m_transporter_registry= 0;
  }
  void setTransporterRegistry(TransporterRegistry *t)
  {
    m_transporter_registry= t;
  }
  SocketServer::Session * newSession(NDB_SOCKET_TYPE socket);
};

class TransporterRegistry : private TransporterSendBufferHandle {
  friend class SHM_Transporter;
  friend class SHM_Writer;
  friend class Transporter;
  friend class TransporterService;
public:
  TransporterRegistry(TransporterCallback *callback,
                      bool use_default_send_buffer = true,
                      unsigned maxTransporters = MAX_NTRANSPORTERS, 
                      unsigned sizeOfLongSignalMemory = 100);

  void set_mgm_handle(NdbMgmHandle h);
  NdbMgmHandle get_mgm_handle(void) { return m_mgm_handle; };

  bool init(NodeId localNodeId);

  bool connect_server(NDB_SOCKET_TYPE sockfd, BaseString& errormsg) const;

  bool connect_client(NdbMgmHandle *h);

  NDB_SOCKET_TYPE connect_ndb_mgmd(SocketClient *sc);

  NDB_SOCKET_TYPE connect_ndb_mgmd(NdbMgmHandle *h);

  void removeAll();
  
  void disconnectAll();

  virtual ~TransporterRegistry();

  bool start_service(SocketServer& server);
  struct NdbThread* start_clients();
  bool stop_clients();
  void start_clients_thread();
  void update_connections();

  void startReceiving();
  void stopReceiving();
  
  void startSending();
  void stopSending();

  // A transporter is always in a PerformState.
  // PerformIO is used initially and as long as any of the events 
  // PerformConnect, ... 
  enum PerformState {
    CONNECTED         = 0,
    CONNECTING        = 1,
    DISCONNECTED      = 2,
    DISCONNECTING     = 3
  };
  const char *getPerformStateString(NodeId nodeId) const
  { return performStateString[(unsigned)performStates[nodeId]]; };

  PerformState getPerformState(NodeId nodeId) const { return performStates[nodeId]; }

  void do_connect(NodeId node_id);
  void do_disconnect(NodeId node_id, int errnum = 0);
  bool is_connected(NodeId node_id) { return performStates[node_id] == CONNECTED; };
  void report_connect(NodeId node_id);
  void report_disconnect(NodeId node_id, int errnum);
  void report_error(NodeId nodeId, TransporterError errorCode,
                    const char *errorInfo = 0);
  
  IOState ioState(NodeId nodeId);
  void setIOState(NodeId nodeId, IOState state);

private:

  bool createTCPTransporter(TransporterConfiguration * config);
  bool createSCITransporter(TransporterConfiguration * config);
  bool createSHMTransporter(TransporterConfiguration * config);

public:
  bool configureTransporter(TransporterConfiguration * config);

  void allocate_send_buffers(Uint64 total_send_buffer);

  Uint64 get_total_max_send_buffer() { return m_total_max_send_buffer; }

  bool get_using_default_send_buffer() const{ return m_use_default_send_buffer;}

  void set_status_overloaded(Uint32 nodeId, bool val);
  const NodeBitmask& get_status_overloaded() const;
  
  SendStatus prepareSend(TransporterSendBufferHandle *sendHandle,
                         const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId, 
                         const LinearSectionPtr ptr[3]);

  SendStatus prepareSend(TransporterSendBufferHandle *sendHandle,
                         const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId, 
                         class SectionSegmentPool & pool,
                         const SegmentedSectionPtr ptr[3]);
  SendStatus prepareSend(TransporterSendBufferHandle *sendHandle,
                         const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId,
                         const GenericSectionPtr ptr[3]);
  SendStatus prepareSend(const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId,
                         const LinearSectionPtr ptr[3])
  {
    return prepareSend(this, signalHeader, prio, signalData, nodeId, ptr);
  }
  SendStatus prepareSend(const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId,
                         class SectionSegmentPool & pool,
                         const SegmentedSectionPtr ptr[3])
  {
    return prepareSend(this, signalHeader, prio, signalData, nodeId, pool, ptr);
  }
  SendStatus prepareSend(const SignalHeader * const signalHeader, Uint8 prio,
                         const Uint32 * const signalData,
                         NodeId nodeId,
                         const GenericSectionPtr ptr[3])
  {
    return prepareSend(this, signalHeader, prio, signalData, nodeId, ptr);
  }
  
  void external_IO(Uint32 timeOutMillis);
  
  inline Uint32 pollReceive(Uint32 timeOutMillis) {
    return pollReceive(timeOutMillis, m_has_data_transporters);
  }
  Uint32 pollReceive(Uint32 timeOutMillis, NodeBitmask& mask);
  void performReceive();
  int performSend(NodeId nodeId);
  void performSend();
  
  int forceSendCheck(int sendLimit);
  
#ifdef DEBUG_TRANSPORTER
  void printState();
#endif

#ifdef ERROR_INSERT
  /* Utils for testing latency issues */
  bool isBlocked(NodeId nodeId);
  void blockReceive(NodeId nodeId);
  void unblockReceive(NodeId nodeId);
#endif
  
  class Transporter_interface {
  public:
    NodeId m_remote_nodeId;
    int m_s_service_port;                       // signed port number
    const char *m_interface;
  };
  Vector<Transporter_interface> m_transporter_interface;
  void add_transporter_interface(NodeId remoteNodeId, const char *interf,
                                 int s_port);   // signed port. <0 is dynamic
  Transporter* get_transporter(NodeId nodeId);
  struct in_addr get_connect_address(NodeId node_id) const;
protected:
  
private:
  TransporterCallback *callbackObj;

  NdbMgmHandle m_mgm_handle;

  struct NdbThread   *m_start_clients_thread;
  bool                m_run_start_clients_thread;

  int sendCounter;
  NodeId localNodeId;
  unsigned maxTransporters;
  int nTransporters;
  int nTCPTransporters;
  int nSCITransporters;
  int nSHMTransporters;

#ifdef ERROR_INSERT
  Bitmask<MAX_NTRANSPORTERS/32> m_blocked;
  Bitmask<MAX_NTRANSPORTERS/32> m_blocked_with_data;
  Bitmask<MAX_NTRANSPORTERS/32> m_blocked_disconnected;
  int m_disconnect_errors[MAX_NTRANSPORTERS];
#endif

  NodeBitmask m_has_data_transporters;
#if defined(HAVE_EPOLL_CREATE)
  int m_epoll_fd;
  struct epoll_event *m_epoll_events;
  bool change_epoll(TCP_Transporter *t, bool add);
#endif

  TCP_Transporter** theTCPTransporters;
  SCI_Transporter** theSCITransporters;
  SHM_Transporter** theSHMTransporters;
  
  TransporterType* theTransporterTypes;
  Transporter**    theTransporters;

  PerformState* performStates;
  int*          m_disconnect_errnum;
  IOState*      ioStates;
  struct ErrorState {
    TransporterError m_code;
    const char *m_info;
  };
  struct ErrorState *m_error_states;

  NodeBitmask m_status_overloaded;
 
  Uint32 unpack(Uint32 * readPtr,
                Uint32 bufferSize,
                NodeId remoteNodeId, 
                IOState state);
  
  Uint32 * unpack(Uint32 * readPtr,
                  Uint32 * eodPtr,
                  NodeId remoteNodeId,
                  IOState state);

  static Uint32 unpack_length_words(const Uint32 *readPtr, Uint32 maxWords);
  void removeTransporter(NodeId nodeId);
  
  int tcpReadSelectReply;
  ndb_socket_poller m_socket_poller;

  Uint32 poll_TCP(Uint32 timeOutMillis, NodeBitmask&);
  Uint32 poll_SCI(Uint32 timeOutMillis, NodeBitmask&);
  Uint32 poll_SHM(Uint32 timeOutMillis, NodeBitmask&);

  int m_shm_own_pid;
  int m_transp_count;

public:
  bool setup_wakeup_socket();
  void wakeup();
private:
  bool m_has_extra_wakeup_socket;
  NDB_SOCKET_TYPE m_extra_wakeup_sockets[2];
  void consume_extra_sockets();


  Uint32 *getWritePtr(TransporterSendBufferHandle *handle,
                      NodeId node, Uint32 lenBytes, Uint32 prio);
  void updateWritePtr(TransporterSendBufferHandle *handle,
                      NodeId node, Uint32 lenBytes, Uint32 prio);

  virtual Uint32 *getWritePtr(NodeId node, Uint32 lenBytes, Uint32 prio,
                              Uint32 max_use);
  virtual Uint32 updateWritePtr(NodeId node, Uint32 lenBytes, Uint32 prio);
  virtual bool forceSend(NodeId node);

private:
  /* Send buffer pages. */
  struct SendBufferPage {
    /* This is the number of words that will fit in one page of send buffer. */
    static const Uint32 PGSIZE = 32768;
    static Uint32 max_data_bytes()
    {
      return PGSIZE - offsetof(SendBufferPage, m_data);
    }

    /* Send buffer for one transporter is kept in a single-linked list. */
    struct SendBufferPage *m_next;

    /* Bytes of send data available in this page. */
    Uint16 m_bytes;
    /* Start of unsent data */
    Uint16 m_start;

    /* Data; real size is to the end of one page. */
    char m_data[2];
  };

  /* Send buffer for one transporter. */
  struct SendBuffer {
    /* Total size of data in buffer, from m_offset_start_data to end. */
    Uint32 m_used_bytes;
    /* Linked list of active buffer pages with first and last pointer. */
    SendBufferPage *m_first_page;
    SendBufferPage *m_last_page;
  };

  SendBufferPage *alloc_page();
  void release_page(SendBufferPage *page);

private:
  /* True if we are using the default send buffer implementation. */
  bool m_use_default_send_buffer;
  /* Send buffers. */
  SendBuffer *m_send_buffers;
  /* Linked list of free pages. */
  SendBufferPage *m_page_freelist;
  /* Original block of memory for pages (so we can free it at exit). */
  unsigned char *m_send_buffer_memory;
  Uint64 m_total_max_send_buffer;

public:
  Uint32 get_bytes_to_send_iovec(NodeId node, struct iovec *dst, Uint32 max);
  Uint32 bytes_sent(NodeId node, Uint32 bytes);
  bool has_data_to_send(NodeId node);

  void reset_send_buffer(NodeId node, bool should_be_empty);

  void print_transporters(const char* where, NdbOut& out = ndbout);

};

inline void
TransporterRegistry::set_status_overloaded(Uint32 nodeId, bool val)
{
  assert(nodeId < MAX_NODES);
  m_status_overloaded.set(nodeId, val);
}

inline const NodeBitmask&
TransporterRegistry::get_status_overloaded() const
{
  return m_status_overloaded;
}

#endif // Define of TransporterRegistry_H