�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fsql%2F'>sql</a>/</p><hr><h3>✏️ Editing: semisync_master_ack_receiver.cc</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>/* Copyright (c) 2014, 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 */

#include &lt;my_global.h&gt;
#include &quot;semisync_master.h&quot;
#include &quot;semisync_master_ack_receiver.h&quot;

#ifdef HAVE_PSI_MUTEX_INTERFACE
extern PSI_mutex_key key_LOCK_ack_receiver;
extern PSI_cond_key key_COND_ack_receiver;
#endif
#ifdef HAVE_PSI_THREAD_INTERFACE
extern PSI_thread_key key_thread_ack_receiver;
#endif

my_socket global_ack_signal_fd= -1;

/* Callback function of ack receive thread */
pthread_handler_t ack_receive_handler(void *arg)
{
  Ack_receiver *recv= reinterpret_cast&lt;Ack_receiver *&gt;(arg);

  my_thread_init();
  my_thread_set_name(&quot;Ack_receiver&quot;);
  recv-&gt;run();
  my_thread_end();

  return NULL;
}

Ack_receiver::Ack_receiver()
{
  DBUG_ENTER(&quot;Ack_receiver::Ack_receiver&quot;);

  m_status= ST_DOWN;
  mysql_mutex_init(key_LOCK_ack_receiver, &amp;m_mutex, NULL);
  mysql_cond_init(key_COND_ack_receiver, &amp;m_cond, NULL);
  mysql_cond_init(key_COND_ack_receiver, &amp;m_cond_reply, NULL);
  m_pid= 0;

  DBUG_VOID_RETURN;
}

void Ack_receiver::cleanup()
{
  DBUG_ENTER(&quot;Ack_receiver::~Ack_receiver&quot;);

  stop();
  mysql_mutex_destroy(&amp;m_mutex);
  mysql_cond_destroy(&amp;m_cond);
  mysql_cond_destroy(&amp;m_cond_reply);

  DBUG_VOID_RETURN;
}

bool Ack_receiver::start()
{
  DBUG_ENTER(&quot;Ack_receiver::start&quot;);

  mysql_mutex_lock(&amp;m_mutex);
  if(m_status == ST_DOWN)
  {
    pthread_attr_t attr;

    m_status= ST_UP;

    if (DBUG_IF(&quot;rpl_semisync_simulate_create_thread_failure&quot;) ||
        pthread_attr_init(&amp;attr) != 0 ||
        pthread_attr_setdetachstate(&amp;attr, PTHREAD_CREATE_JOINABLE) != 0 ||
#ifndef _WIN32
        pthread_attr_setscope(&amp;attr, PTHREAD_SCOPE_SYSTEM) != 0 ||
#endif
        mysql_thread_create(key_thread_ack_receiver, &amp;m_pid,
                            &amp;attr, ack_receive_handler, this))
    {
      sql_print_error(&quot;Failed to start semi-sync ACK receiver thread, &quot;
                      &quot; could not create thread(errno:%d)&quot;, errno);

      m_status= ST_DOWN;
      mysql_mutex_unlock(&amp;m_mutex);

      DBUG_RETURN(true);
    }
    (void) pthread_attr_destroy(&amp;attr);
  }
  mysql_mutex_unlock(&amp;m_mutex);

  DBUG_RETURN(false);
}

void Ack_receiver::stop()
{
  DBUG_ENTER(&quot;Ack_receiver::stop&quot;);

  mysql_mutex_lock(&amp;m_mutex);
  if (m_status == ST_UP)
  {
    m_status= ST_STOPPING;
    signal_listener();            // Signal listener thread to stop
    mysql_cond_broadcast(&amp;m_cond);

    while (m_status == ST_STOPPING)
      mysql_cond_wait(&amp;m_cond, &amp;m_mutex);

    DBUG_ASSERT(m_status == ST_DOWN);

    m_pid= 0;
  }
  mysql_mutex_unlock(&amp;m_mutex);

  DBUG_VOID_RETURN;
}

#ifndef DBUG_OFF
void static dbug_verify_no_duplicate_slaves(Slave_ilist *m_slaves, THD *thd)
{
  I_List_iterator&lt;Slave&gt; it(*m_slaves);
  Slave *slave;
  while ((slave= it++))
  {
    DBUG_ASSERT(slave-&gt;thd-&gt;variables.server_id != thd-&gt;variables.server_id);
  }
}
#else
#define dbug_verify_no_duplicate_slaves(A,B) do {} while(0)
#endif


bool Ack_receiver::add_slave(THD *thd)
{
  Slave *slave;
  DBUG_ENTER(&quot;Ack_receiver::add_slave&quot;);

  if (!(slave= new Slave))
    DBUG_RETURN(true);

  slave-&gt;active= 0;
  slave-&gt;thd= thd;
  slave-&gt;vio= *thd-&gt;net.vio;
  slave-&gt;vio.mysql_socket.m_psi= NULL;
  slave-&gt;vio.read_timeout= 1;                   // 1 ms

  mysql_mutex_lock(&amp;m_mutex);

  dbug_verify_no_duplicate_slaves(&amp;m_slaves, thd);

  m_slaves.push_back(slave);
  m_slaves_changed= true;
  mysql_cond_broadcast(&amp;m_cond);
  mysql_mutex_unlock(&amp;m_mutex);

  signal_listener();  // Inform listener that there are new slaves

  DBUG_RETURN(false);
}

void Ack_receiver::remove_slave(THD *thd)
{
  I_List_iterator&lt;Slave&gt; it(m_slaves);
  Slave *slave;
  bool slaves_changed= 0;
  DBUG_ENTER(&quot;Ack_receiver::remove_slave&quot;);

  mysql_mutex_lock(&amp;m_mutex);

  while ((slave= it++))
  {
    if (slave-&gt;thd == thd)
    {
      delete slave;
      slaves_changed= true;
      break;
    }
  }
  if (slaves_changed)
  {
    m_slaves_changed= true;
    mysql_cond_broadcast(&amp;m_cond);
    /*
      Wait until Ack_receiver::run() acknowledges remove of slave
      As this is only sent under the mutex and after listners has
      been collected, we know that listener has ignored the found
      slave.
    */
    if (m_status != ST_DOWN)
      mysql_cond_wait(&amp;m_cond_reply, &amp;m_mutex);
  }
  mysql_mutex_unlock(&amp;m_mutex);

  DBUG_VOID_RETURN;
}

inline void Ack_receiver::set_stage_info(const PSI_stage_info &amp;stage)
{
  (void)MYSQL_SET_STAGE(stage.m_key, __FILE__, __LINE__);
}

void Ack_receiver::wait_for_slave_connection(THD *thd)
{
  thd-&gt;enter_cond(&amp;m_cond, &amp;m_mutex, &amp;stage_waiting_for_semi_sync_slave,
                  0, __func__, __FILE__, __LINE__);

  while (m_status == ST_UP &amp;&amp; m_slaves.is_empty())
    mysql_cond_wait(&amp;m_cond, &amp;m_mutex);

  thd-&gt;exit_cond(0, __func__, __FILE__, __LINE__);
}

/* Auxilary function to initialize a NET object with given net buffer. */
static void init_net(NET *net, unsigned char *buff, unsigned int buff_len)
{
  memset(net, 0, sizeof(NET));
  net-&gt;max_packet= buff_len;
  net-&gt;buff= buff;
  net-&gt;buff_end= buff + buff_len;
  net-&gt;read_pos= net-&gt;buff;
}

void Ack_receiver::run()
{
  THD *thd= new THD(next_thread_id());
  NET net;
  unsigned char net_buff[REPLY_MESSAGE_MAX_LENGTH];
  DBUG_ENTER(&quot;Ack_receiver::run&quot;);

  my_thread_init();

#ifdef HAVE_POLL
  Poll_socket_listener listener(m_slaves);
#else
  Select_socket_listener listener(m_slaves);
#endif //HAVE_POLL

  if (listener.got_error())
  {
    sql_print_error(&quot;Got error %M starting ack receiver thread&quot;,
                    listener.got_error());
    return;
  }

  sql_print_information(&quot;Starting ack receiver thread&quot;);
  thd-&gt;system_thread= SYSTEM_THREAD_SEMISYNC_MASTER_BACKGROUND;
  thd-&gt;store_globals();
  thd-&gt;security_ctx-&gt;skip_grants();
  thd-&gt;set_command(COM_DAEMON);
  init_net(&amp;net, net_buff, REPLY_MESSAGE_MAX_LENGTH);

  /*
    Mark that we have to setup the listener. Note that only this functions can
    set m_slaves_changed to false
  */
  m_slaves_changed= true;

  while (1)
  {
    int ret, slave_count= 0;
    Slave *slave;

    mysql_mutex_lock(&amp;m_mutex);
    if (unlikely(m_status != ST_UP))
      goto end;

    if (unlikely(m_slaves_changed))
    {
      if (unlikely(m_slaves.is_empty()))
      {
        m_slaves_changed= false;
        mysql_cond_broadcast(&amp;m_cond_reply);      // Signal remove_slave
        wait_for_slave_connection(thd);
        /* Wait for slave unlocks m_mutex */
        continue;
      }

      set_stage_info(stage_waiting_for_semi_sync_ack_from_slave);
      if ((slave_count= listener.init_slave_sockets()) == 0)
      {
        mysql_mutex_unlock(&amp;m_mutex);
        m_slaves_changed= true;
        continue;                               // Retry
      }
      if (slave_count &lt; 0)
        goto end;
      m_slaves_changed= false;
      mysql_cond_broadcast(&amp;m_cond_reply);      // Signal remove_slave
    }

#ifdef HAVE_POLL
      DBUG_PRINT(&quot;info&quot;, (&quot;fd count %u&quot;, slave_count));
#else     
      DBUG_PRINT(&quot;info&quot;, (&quot;fd count %u, max_fd %d&quot;, slave_count,
                          (int) listener.get_max_fd()));
#endif

    mysql_mutex_unlock(&amp;m_mutex);
    ret= listener.listen_on_sockets();

    if (ret &lt;= 0)
    {

      ret= DBUG_IF(&quot;rpl_semisync_simulate_select_error&quot;) ? -1 : ret;

      if (ret == -1 &amp;&amp; errno != EINTR)
        sql_print_information(&quot;Failed to wait on semi-sync sockets, &quot;
                              &quot;error: errno=%d&quot;, socket_errno);
      continue;
    }

    listener.clear_signal();
    mysql_mutex_lock(&amp;m_mutex);
    set_stage_info(stage_reading_semi_sync_ack);
    Slave_ilist_iterator it(m_slaves);
    while ((slave= it++))
    {
      if (slave-&gt;active &amp;&amp;
          ((slave-&gt;vio.read_pos &lt; slave-&gt;vio.read_end) ||
           listener.is_socket_active(slave)))
      {
        ulong len;

        /* Semi-sync packets will always be sent with pkt_nr == 1 */
        net_clear(&amp;net, 0);
        net.vio= &amp;slave-&gt;vio;
        /*
          Set compress flag. This is needed to support
          Slave_compress_protocol flag enabled Slaves
        */
        net.compress= slave-&gt;thd-&gt;net.compress;

        if (unlikely(listener.is_socket_hangup(slave)))
        {
          if (global_system_variables.log_warnings &gt; 2)
            sql_print_warning(&quot;Semisync ack receiver got hangup &quot;
                              &quot;from slave server-id %d&quot;,
                              slave-&gt;server_id());
          it.remove();
          m_slaves_changed= true;
          continue;
        }

        len= my_net_read(&amp;net);
        if (likely(len != packet_error))
        {
          int res;
          res= repl_semisync_master.report_reply_packet(slave-&gt;server_id(),
                                                        net.read_pos, len);
          if (unlikely(res &lt; 0))
          {
            /*
              Slave has sent COM_QUIT or other failure.
              Delete it from listener
            */
            it.remove();
            m_slaves_changed= true;
          }
        }
        else if (net.last_errno == ER_NET_READ_ERROR)
        {
          if (net.last_errno &gt; 0 &amp;&amp; global_system_variables.log_warnings &gt; 2)
            sql_print_warning(&quot;Semisync ack receiver got error %d \&quot;%s\&quot; &quot;
                              &quot;from slave server-id %d&quot;,
                              net.last_errno, ER_DEFAULT(net.last_errno),
                              slave-&gt;server_id());
          it.remove();
          m_slaves_changed= true;
        }
      }
    }
    mysql_mutex_unlock(&amp;m_mutex);
  }

end:
  sql_print_information(&quot;Stopping ack receiver thread&quot;);
  m_status= ST_DOWN;
  mysql_cond_broadcast(&amp;m_cond);
  mysql_cond_broadcast(&amp;m_cond_reply);
  mysql_mutex_unlock(&amp;m_mutex);

  delete thd;
  DBUG_VOID_RETURN;
}
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fsql'>🔙 Back</a></p>