Below is the list of changes that have just been committed into a local
5.1 repository of jonas. When jonas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html
ChangeSet@stripped, 2007-05-21 16:47:15+02:00, jonas@stripped +1 -0
ndb - add new interface for thread safe signal packing
storage/ndb/src/common/transporter/thr_client_r.cpp@stripped, 2007-05-21 16:47:12+02:00, jonas@stripped +248 -0
New BitKeeper file ``storage/ndb/src/common/transporter/thr_client_r.cpp''
storage/ndb/src/common/transporter/thr_client_r.cpp@stripped, 2007-05-21 16:47:12+02:00, jonas@stripped +0 -0
# This is a BitKeeper patch. What follows are the unified diffs for the
# set of deltas contained in the patch. The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User: jonas
# Host: eel.hemma.oreland.se
# Root: /home/jonas/src/51-telco
--- New file ---
+++ storage/ndb/src/common/transporter/thr_client_r.cpp 07/05/21 16:47:12
/* Copyright (C) 2003 MySQL AB
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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "TransporterRegistry.hpp"
#include "Transporter.hpp"
#include "TCP_Transporter.hpp"
struct thr_client_r
{
class Transporter * m_transporter;
Uint32 m_node_id, m_bytes, m_cnt;
struct iovec m_buffers[1];
};
thr_client_r*
TransporterRegistry::create_client(NodeId nodeId)
{
const Uint32 SZ = NDB_TRANSPORTER_PAGE_SIZE;
/**
* Compute m_sz
*/
Uint32 bufsz = m_send_buffer_size;
// Convert to pages
bufsz = (bufsz + SZ - 1) / SZ;
// dimension size of m_buffers as power of 2
Uint32 n = 0;
while(bufsz) { n++; bufsz /= 2; }
if (n < 16)
n = 16;
void * ptr = malloc(sizeof(thr_client_r) + (n - 1) * sizeof(struct iovec));
if (ptr == 0)
{
return 0;
}
bzero(ptr, sizeof(thr_client_r));
thr_client_r* ret = (thr_client_r*)ptr;
if (nodeId)
{
if (reset_node(ret, nodeId) == 0)
return ret;
free(ptr);
return 0;
}
return ret;
}
void
TransporterRegistry::destroy_client(struct thr_client_r * thr)
{
const Uint32 SZ = NDB_TRANSPORTER_PAGE_SIZE;
Uint32 cnt = (thr->m_bytes + SZ - 1) / SZ;
release_buffers(thr->m_buffers, cnt);
free(thr);
}
int
TransporterRegistry::reset_node(struct thr_client_r * thr, Uint32 node)
{
Uint32 bytes = thr->m_bytes;
Uint32 curr_node = thr->m_node_id;
if (unlikely(! (bytes == 0 &&
curr_node == 0 &&
node < maxTransporters)))
{
assert(false);
return -1;
}
thr->m_node_id = node;
thr->m_transporter = theTransporters[node];
return 0;
}
Uint32
TransporterRegistry::get_buffer(struct thr_client_r* thr,
Uint32 sz, Uint32* dst[2])
{
Uint32 bytes = thr->m_bytes;
Uint32 idx = bytes / NDB_TRANSPORTER_PAGE_SIZE;
Uint32 cnt = thr->m_cnt;
struct iovec *bufs = thr->m_buffers;
struct iovec *buf = bufs + idx;
Uint32 pos = buf->iov_len;
dst[0] = (Uint32*)((char*)buf->iov_base + pos);
thr->m_bytes = bytes + sz;
if (pos + sz < NDB_TRANSPORTER_PAGE_SIZE)
{
buf->iov_len = pos + sz;
return sz;
}
/**
* Signal did not fit current page (or made it 100% full)
*/
buf->iov_len = NDB_TRANSPORTER_PAGE_SIZE;
buf = bufs + idx + 1;
if (unlikely(idx + 1 == cnt))
{
thr->m_bytes = bytes;
return 0; // buffer full
}
Uint32* ptr = alloc_buffer();
if (unlikely(ptr == 0))
{
thr->m_bytes = bytes;
return 0;
}
bufs->iov_len = 0;
bufs->iov_base = ptr;
if (pos + sz == NDB_TRANSPORTER_PAGE_SIZE)
{
return sz;
}
dst[1] = ptr;
return NDB_TRANSPORTER_PAGE_SIZE - pos;
}
SendStatus
TransporterRegistry::prepareSend(struct thr_client_r * thr,
const SignalHeader * const header,
Uint8 prio,
const Uint32 * const theData,
const LinearSectionPtr ptr[3])
{
Uint32 ret;
Packer * pck = &thr->m_transporter->m_packer;
Uint32 sz = pck->getMessageLength(header, ptr);
Uint32 * dst[2];
if (likely((ret = get_buffer(thr, sz, dst)) == sz))
{
pck->pack(dst[0], prio, header, theData, ptr);
return SEND_OK;
}
else if (ret)
{
pck->pack(dst[1], prio, header, theData, ptr);
memcpy(dst[0], dst[1], ret);
memmove(dst[1], ((char*)dst[1])+ret, sz - ret);
return SEND_OK;
}
else
{
return SEND_BUFFER_FULL;
}
}
SendStatus
TransporterRegistry::flush(struct thr_client_r* thr)
{
Uint32 bytes = thr->m_bytes;
struct iovec * bufs = thr->m_buffers;
SendStatus ret = thr->m_transporter->prepareSend(this, bufs, bytes);
if (ret == SEND_OK)
{
thr->m_bytes = 0;
return SEND_OK;
}
return ret;
}
bool
TransporterRegistry::alloc_buffers()
{
m_send_buffer_size &= ~(Uint32)(NDB_TRANSPORTER_PAGE_SIZE - 1);
void * tmp = malloc(m_send_buffer_size);
if (tmp == 0)
return false;
Uint32 * ptr = (Uint32*)tmp;
Uint32 cnt = m_send_buffer_size / NDB_TRANSPORTER_PAGE_SIZE;
m_first_free_buffer = 0;
while(cnt)
{
struct iovec buf;
buf.iov_base = ptr;
release_buffers(&buf, 1);
ptr += (NDB_TRANSPORTER_PAGE_SIZE / 4);
cnt--;
}
for (int i = 0; i<nTCPTransporters; i++)
{
Uint32 * ptr = alloc_buffer();
theTCPTransporters[i]->m_send_buffers[0].iov_base = ptr;
}
return true;
}
Uint32*
TransporterRegistry::alloc_buffer()
{
Uint32 * ptr = m_first_free_buffer;
if (ptr)
{
Uint32 ** next = reinterpret_cast<Uint32**>(ptr);
m_first_free_buffer = *next;
return ptr;
}
return 0;
}
void
TransporterRegistry::release_buffers(struct iovec* bufs, Uint32 n)
{
Uint32* ff = m_first_free_buffer;
for (Uint32 i = 0; i<n; i++)
{
Uint32* ptr = (Uint32*)bufs[i].iov_base;
Uint32** next = (Uint32**)ptr;
* next = ff;
m_first_free_buffer = ptr;
}
}
| Thread |
|---|
| • bk commit into 5.1 tree (jonas:1.2525) | jonas | 29 May |
