table_session_connect.cc

/* Copyright (c) 2008, 2012, 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,
  51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA */

#include "table_session_connect.h"

static const TABLE_FIELD_TYPE field_types[]=
{
  {
    { C_STRING_WITH_LEN("PROCESSLIST_ID") },
    { C_STRING_WITH_LEN("int(11)") },
    { NULL, 0}
  },
  {
    { C_STRING_WITH_LEN("ATTR_NAME") },
    { C_STRING_WITH_LEN("varchar(32)") },
    { NULL, 0}
  },
  {
    { C_STRING_WITH_LEN("ATTR_VALUE") },
    { C_STRING_WITH_LEN("varchar(1024)") },
    { NULL, 0}
  },
  {
    { C_STRING_WITH_LEN("ORDINAL_POSITION") },
    { C_STRING_WITH_LEN("int(11)") },
    { NULL, 0}
  }
};

TABLE_FIELD_DEF table_session_connect::m_field_def=
{ 4, field_types };

table_session_connect::table_session_connect(const PFS_engine_table_share *share) :
                       cursor_by_thread_connect_attr(share)
{}

bool parse_length_encoded_string(const char **ptr,
                 char *dest, uint dest_size,
                 uint *copied_len,
                 const char *start_ptr, uint input_length,
                 bool copy_data,
                 const CHARSET_INFO *from_cs,
                 uint nchars_max)
{
  ulong copy_length, data_length;
  const char *well_formed_error_pos= NULL, *cannot_convert_error_pos= NULL,
        *from_end_pos= NULL;

  copy_length= data_length= net_field_length((uchar **) ptr);

  /* we don't tolerate NULL as a length */
  if (data_length == NULL_LENGTH)
    return true;

  if (*ptr - start_ptr + data_length > input_length)
    return true;

  copy_length= well_formed_copy_nchars(&my_charset_utf8_bin, dest, dest_size,
                                       from_cs, *ptr, data_length, nchars_max,
                                       &well_formed_error_pos,
                                       &cannot_convert_error_pos,
                                       &from_end_pos);
  *copied_len= copy_length;
  (*ptr)+= data_length;

  return false;
}

bool read_nth_attr(const char *connect_attrs,
                   uint connect_attrs_length,
                   const CHARSET_INFO *connect_attrs_cs,
                   uint ordinal,
                   char *attr_name, uint max_attr_name,
                   uint *attr_name_length,
                   char *attr_value, uint max_attr_value,
                   uint *attr_value_length)
{
  uint idx;
  const char *ptr;

  for (ptr= connect_attrs, idx= 0;
       (uint)(ptr - connect_attrs) < connect_attrs_length && idx <= ordinal;
      idx++)
  {
    uint copy_length;
    /* do the copying only if we absolutely have to */
    bool fill_in_attr_name= idx == ordinal;
    bool fill_in_attr_value= idx == ordinal;

    /* read the key */
    if (parse_length_encoded_string(&ptr,
                                    attr_name, max_attr_name, &copy_length,
                                    connect_attrs,
                                    connect_attrs_length,
                                    fill_in_attr_name,
                                    connect_attrs_cs, 32) ||
        !copy_length
        )
      return false;

    if (idx == ordinal)
      *attr_name_length= copy_length;
      
    /* read the value */
    if (parse_length_encoded_string(&ptr,
                                    attr_value, max_attr_value, &copy_length,
                                    connect_attrs,
                                    connect_attrs_length,
                                    fill_in_attr_value,
                                    connect_attrs_cs, 1024))
      return false;

    if (idx == ordinal)
      *attr_value_length= copy_length;

    if (idx == ordinal)
      return true;
  }

  return false;
}

void table_session_connect::make_row(PFS_thread *pfs, uint ordinal)
{
  pfs_lock lock;
  PFS_thread_class *safe_class;

  m_row_exists= false;

  /* Protect this reader against thread termination */
  pfs->m_lock.begin_optimistic_lock(&lock);
  safe_class= sanitize_thread_class(pfs->m_class);
  if (unlikely(safe_class == NULL))
    return;

  /* Filtering threads must be done under the protection of the optimistic lock. */
  if (! thread_fits(pfs))
    return;

  /* populate the row */
  if (read_nth_attr(pfs->m_session_connect_attrs,
                    pfs->m_session_connect_attrs_length,
                    pfs->m_session_connect_attrs_cs,
                    ordinal,
                    m_row.m_attr_name, (uint) sizeof(m_row.m_attr_name),
                    &m_row.m_attr_name_length,
                    m_row.m_attr_value, (uint) sizeof(m_row.m_attr_value),
                    &m_row.m_attr_value_length))
  {
    /* we don't expect internal threads to have connection attributes */
    DBUG_ASSERT(pfs->m_processlist_id != 0);

    m_row.m_ordinal_position= ordinal;
    m_row.m_process_id= pfs->m_processlist_id;
  }
  else
    return;

  if (pfs->m_lock.end_optimistic_lock(& lock))
    m_row_exists= true;
}

int table_session_connect::read_row_values(TABLE *table,
                                         unsigned char *buf,
                                         Field **fields,
                                         bool read_all)
{
  Field *f;

  if (unlikely(!m_row_exists))
    return HA_ERR_RECORD_DELETED;

  /* Set the null bits */
  DBUG_ASSERT(table->s->null_bytes == 1);
  buf[0]= 0;

  for (; (f= *fields) ; fields++)
  {
    if (read_all || bitmap_is_set(table->read_set, f->field_index))
    {
      switch(f->field_index)
      {
      case FO_PROCESS_ID:
        if (m_row.m_process_id != 0)
          set_field_ulong(f, m_row.m_process_id);
        else
          f->set_null();
        break;
      case FO_ATTR_NAME:
        set_field_varchar_utf8(f, m_row.m_attr_name,
                               m_row.m_attr_name_length);
        break;
      case FO_ATTR_VALUE:
        if (m_row.m_attr_value_length)
          set_field_varchar_utf8(f, m_row.m_attr_value,
                                 m_row.m_attr_value_length);
        else
          f->set_null();
        break;
      case FO_ORDINAL_POSITION:
        set_field_ulong(f, m_row.m_ordinal_position);
        break;
      default:
        DBUG_ASSERT(false);
      }
    }
  }
  return 0;
}

bool
table_session_connect::thread_fits(PFS_thread *thread)
{
  return true;
}