context: /*
      Copyright (c) 2007-2012 iMatix Corporation
      Copyright (c) 2009-2011 250bpm s.r.o.
      Copyright (c) 2007-2015 Other contributors as noted in the AUTHORS file
  
      This file is part of 0MQ.
  
      0MQ is free software; you can redistribute it and/or modify it under
      the terms of the GNU Lesser General Public License as published by
      the Free Software Foundation; either version 3 of the License, or
      (at your option) any later version.
  
      0MQ 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 Lesser General Public License for more details.
  
      You should have received a copy of the GNU Lesser General Public License
      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
  
  using System;
  using System.Collections.Generic;
  using System.Diagnostics;
  using System.Linq;
  using System.Threading;
  using JetBrains.Annotations;
  
  namespace NetMQ.Core
  {
      /// <summary>
      /// Objects of class Ctx are intended to encapsulate all of the global state
      /// associated with the NetMQ library. This contains the sockets, and manages interaction
      /// between them.
      /// </summary>
      /// <remarks>Internal analog of the public <see cref="NetMQContext"/> class.</remarks>
      internal sealed class Ctx
      {
          internal const int DefaultIOThreads = 1;
          internal const int DefaultMaxSockets = 1024;
  
          #region Nested class: Endpoint
  
          /// <summary>
          /// Information associated with inproc endpoint. Note that endpoint options
          /// are registered as well so that the peer can access them without a need
          /// for synchronisation, handshaking or similar.
          /// </summary>
          public class Endpoint
          {
              /// <summary>
              /// Create a new Endpoint with the given socket.
              /// </summary>
              /// <param name="socket">the socket for this new Endpoint</param>
              /// <param name="options">the Options to assign to this new Endpoint</param>
              public Endpoint([NotNull] SocketBase socket, [NotNull] Options options)
              {
                  Socket = socket;
                  Options = options;
              }
  
              /// <summary>
              /// Get the socket associated with this Endpoint.
              /// </summary>
              [NotNull]
              public SocketBase Socket { get; private set; }
  
              /// <summary>
              /// Get the Options of this Endpoint.
              /// </summary>
              [NotNull]
              public Options Options { get; private set; }
          }
  
          #endregion
  
          private bool m_disposed;
  
          /// <summary>
          /// Sockets belonging to this context. We need the list so that
          /// we can notify the sockets when zmq_term() is called. The sockets
          /// will return ETERM then.
          /// </summary>
          private readonly List<SocketBase> m_sockets = new List<SocketBase>();
  
          /// <summary>
          /// List of unused thread slots.
          /// </summary>
          private readonly Stack<int> m_emptySlots = new Stack<int>();
  
          /// <summary>
          /// If true, zmq_init has been called but no socket has been created
          /// yet. Launching of I/O threads is delayed.
          /// </summary>
          private volatile bool m_starting = true;
  
          /// <summary>
          /// If true, zmq_term was already called.
          /// </summary>
          private bool m_terminating;
  
          /// <summary>
          /// This object is for synchronisation of accesses to global slot-related data:
          /// sockets, empty_slots, terminating. It also synchronises
          /// access to zombie sockets as such (as opposed to slots) and provides
          /// a memory barrier to ensure that all CPU cores see the same data.
          /// </summary>
          private readonly object m_slotSync = new object();
  
          /// <summary>
          /// The reaper thread.
          /// </summary>
          [CanBeNull] private Reaper m_reaper;
  
          /// <summary>
          /// List of I/O threads.
          /// </summary>
          private readonly List<IOThread> m_ioThreads = new List<IOThread>();
  
          /// <summary>
          /// Length of the mailbox-array.
          /// </summary>
          private int m_slotCount;
  
          /// <summary>
          /// Array of pointers to mailboxes for both application and I/O threads.
          /// </summary>
          [CanBeNull] private IMailbox[] m_slots;
  
          /// <summary>
          /// Mailbox for zmq_term thread.
          /// </summary>
          private readonly Mailbox m_termMailbox = new Mailbox("terminator");
  
          /// <summary>
          /// Dictionary containing the inproc endpoints within this context.
          /// </summary>
          private readonly Dictionary<string, Endpoint> m_endpoints = new Dictionary<string, Endpoint>();
  
          /// <summary>
          /// This object provides synchronisation of access to the list of inproc endpoints.
          /// </summary>
          private readonly object m_endpointsSync = new object();
  
          /// <summary>
          /// The highest socket-id that has been assigned thus far.
          /// </summary>
          private static int s_maxSocketId;
  
          /// <summary>
          /// The maximum number of sockets that can be opened at the same time.
          /// </summary>
          private int m_maxSockets = DefaultMaxSockets;
  
          /// <summary>
          /// The number of I/O threads to launch.
          /// </summary>
          private int m_ioThreadCount = DefaultIOThreads;
  
          /// <summary>
          /// Should the context termination block until all socket are disposed?
          /// </summary>
          private bool m_block = true;
  
          /// <summary>
          /// This object is used to synchronize access to context options.
          /// </summary>
          private readonly object m_optSync = new object();
  
          /// <summary>
          /// The thread-id for the termination (the equivalent of the zmq_term) thread.
          /// </summary>
          public const int TermTid = 0;
  
          /// <summary>
          /// This is the thread-id to assign to the Reaper (value is 1).
          /// </summary>
          public const int ReaperTid = 1;
          
          /// <summary>Throws <see cref="ObjectDisposedException"/> if this is already disposed.</summary>
          /// <exception cref="ObjectDisposedException">This object has already been disposed.</exception>
          public void CheckDisposed()
          {
              if (m_disposed)
                  throw new ObjectDisposedException(GetType().FullName);
          }
  
          /// <summary>
          /// This function is called when user invokes zmq_term. If there are
          /// no more sockets open it'll cause all the infrastructure to be shut
          /// down. If there are open sockets still, the deallocation happens
          /// after the last one is closed.
          /// </summary>
          public void Terminate()
          {
              m_disposed = true;
  
              Monitor.Enter(m_slotSync);
  
              if (!m_starting)
              {
                  // Check whether termination was already underway, but interrupted and now
                  // restarted.
                  bool restarted = m_terminating;
                  m_terminating = true;
                  Monitor.Exit(m_slotSync);
  
                  // First attempt to terminate the context.
                  if (!restarted)
                  {
                      // First send stop command to sockets so that any blocking calls
                      // can be interrupted. If there are no sockets we can ask reaper
                      // thread to stop.
                      Monitor.Enter(m_slotSync);
                      try
                      {
                          foreach (var socket in m_sockets)
                              socket.Stop();
  
                          if (!Block)
                          {
                              m_reaper.ForceStop();
                          }
                          else if (m_sockets.Count == 0)
                              m_reaper.Stop();                        
                      }
                      finally
                      {
                          Monitor.Exit(m_slotSync);
                      }
                  }
  
                  if (m_block)
                  {
                      // Wait till reaper thread closes all the sockets.
                      Command command;
                      var found = m_termMailbox.TryRecv(-1, out command);
  
                      Debug.Assert(found);
                      Debug.Assert(command.CommandType == CommandType.Done);
                      Monitor.Enter(m_slotSync);                
                  }
                  else
                      Monitor.Enter(m_slotSync);
              }
              Monitor.Exit(m_slotSync);
  
              // Deallocate the resources.
              foreach (IOThread it in m_ioThreads)
                  it.Stop();
  
              foreach (IOThread it in m_ioThreads)
                  it.Destroy();
  
              if (m_reaper != null)
                  m_reaper.Destroy();
  
              m_termMailbox.Close();
  
              m_disposed = true;
          }
  
          public int IOThreadCount
          {
              get { return m_ioThreadCount; }
              set
              {
                  if (value < 0)
                      throw new ArgumentOutOfRangeException("value", value, "Must be zero or greater");
                  lock (m_optSync)
                      m_ioThreadCount = value;
              }
          }
  
          public int MaxSockets
          {
              get { return m_maxSockets; }
              set
              {
                  if (value <= 0)
                      throw new ArgumentOutOfRangeException("value", value, "Must be greater than zero");
                  lock (m_optSync)
                      m_maxSockets = value;
              }
          }
  
          /// <summary>
          /// Should context wait for all sockets before terminating?
          /// </summary>
          public bool Block
          {
  
              get { return m_block; }
              set
              {                
                  lock (m_optSync)
                      m_block = value;
              }
          }
  
          /// <summary>
          /// Create and return a new socket of the given type, and initialize this Ctx if this is the first one.
          /// </summary>
          /// <param name="type">the type of socket to create</param>
          /// <returns>the newly-created socket</returns>
          /// <exception cref="TerminatingException">Cannot create new socket while terminating.</exception>
          /// <exception cref="NetMQException">Maximum number of sockets reached.</exception>
          /// <exception cref="TerminatingException">The context (Ctx) must not be already terminating.</exception>
          [NotNull]
          public SocketBase CreateSocket(ZmqSocketType type)
          {
              lock (m_slotSync)
              {
                  if (m_starting)
                  {
                      m_starting = false;
                      // Initialise the array of mailboxes. Additional three slots are for
                      // zmq_term thread and reaper thread.
  
                      int ios;
                      int mazmq;
  
                      lock (m_optSync)
                      {
                          mazmq = m_maxSockets;
                          ios = m_ioThreadCount;
                      }
                      m_slotCount = mazmq + ios + 2;
                      m_slots = new IMailbox[m_slotCount];
                      //alloc_Debug.Assert(slots);
  
                      // Initialise the infrastructure for zmq_term thread.
                      m_slots[TermTid] = m_termMailbox;
  
                      // Create the reaper thread.
                      m_reaper = new Reaper(this, ReaperTid);
                      //alloc_Debug.Assert(reaper);
                      m_slots[ReaperTid] = m_reaper.Mailbox;
                      m_reaper.Start();
  
                      // Create I/O thread objects and launch them.
                      for (int i = 2; i != ios + 2; i++)
                      {
                          var ioThread = new IOThread(this, i);
                          //alloc_Debug.Assert(io_thread);
                          m_ioThreads.Add(ioThread);
                          m_slots[i] = ioThread.Mailbox;
                          ioThread.Start();
                      }
  
                      // In the unused part of the slot array, create a list of empty slots.
                      for (int i = m_slotCount - 1; i >= ios + 2; i--)
                      {
                          m_emptySlots.Push(i);
                          m_slots[i] = null;
                      }
                  }
  
                  // Once zmq_term() was called, we can't create new sockets.
                  if (m_terminating)
                  {
                      string xMsg = string.Format("Ctx.CreateSocket({0}), cannot create new socket while terminating.", type);
                      throw new TerminatingException(innerException: null, message: xMsg);
                  }
  
                  // If max_sockets limit was reached, return error.
                  if (m_emptySlots.Count == 0)
                  {
  #if DEBUG
                      string xMsg = string.Format("Ctx.CreateSocket({0}), max number of sockets {1} reached.", type, m_maxSockets);
                      throw NetMQException.Create(xMsg, ErrorCode.TooManyOpenSockets);
  #else
                      throw NetMQException.Create(ErrorCode.TooManyOpenSockets);
  #endif
                  }
  
                  // Choose a slot for the socket.
                  int slot = m_emptySlots.Pop();
  
                  // Generate new unique socket ID.
                  int socketId = Interlocked.Increment(ref s_maxSocketId);
  
                  // Create the socket and register its mailbox.
                  SocketBase s = SocketBase.Create(type, this, slot, socketId);
  
                  m_sockets.Add(s);
  
gt:                 m_slots[slot] = s.Mailbox;