Hi all,
Here's the updated version. It can work on a 2-vmware-nodes hb2 cluster
with
several dummy resources running for hours now. :)
2007/6/1, Lars Marowsky-Bree <[EMAIL PROTECTED]>:
On 2007-06-01T17:22:24, Xinwei Hu <[EMAIL PROTECTED]> wrote:
Random comments, which I've not sorted:
Implement a circular buffer (one per node) for writes, with an
additional meta data section (one slot per node).
Circular buffers are used in current version.
You can atomically update the pointer as to where the buffer currently
begins and ends - sector-sized writes (512b, on SAN arrays it's often
higher) are atomic. The slots should contain at least the uname, UUID,
liveness counter, deadtime, keepalive interval, cluster name.
Will try in next version.
A static sized buffer of a few MB per node shouldn't be too hard to
implement. Adding slots at the end shouldn't either. Resizing the
buffers dynamically is going to be hard and might not be needed.
Possible optimization: Filter out pure heartbeat packets - write the
liveness information into a counter in the slot instead, the reading
nodes can then infer the heartbeats by seeing it increase. (This is
advanced, but may be a very good idea to be able to do in the MCP anyway
- otherwise, the disk heads will be constantly moving very fast.)
Will try in next version. :)
Why do you need SHA inside the plugin!? The packets you receive are
already authenticated and signed.
I use multi-blocks for each message. In case of half-written blocks, I use
sha to compute checksum.
You don't need to authenticate the slot itself, I'd argue. 512b writes
are atomic - you're not going to see a half-way written one.
Regards,
Lars
--
Teamlead Kernel, SuSE Labs, Research and Development
SUSE LINUX Products GmbH, GF: Markus Rex, HRB 16746 (AG Nürnberg)
"Experience is the name everyone gives to their mistakes." -- Oscar Wilde
_______________________________________________________
Linux-HA-Dev: [email protected]
http://lists.linux-ha.org/mailman/listinfo/linux-ha-dev
Home Page: http://linux-ha.org/
/*
* dskcm.c: Disk based communication code for heartbeat.
*
* Copyright (C) 2007 Xinwei Hu <[EMAIL PROTECTED]>
*
* This library 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 2.1 of the License, or (at your option) any later version.
*
* This library 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 library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
*/
#define _GNU_SOURCE
#define _XOPEN_SOURCE 600
#define _POSIX_C_SOURCE 199309
#ifndef DSKCM_STANDALONE
#include <lha_internal.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <ctype.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <time.h>
#ifndef DSKCM_STANDALONE
#include <heartbeat.h>
#include <HBcomm.h>
#define PIL_PLUGINTYPE HB_COMM_TYPE
#define PIL_PLUGINTYPE_S HB_COMM_TYPE_S
#define PIL_PLUGIN dskcm
#define PIL_PLUGIN_S "dskcm"
#define PIL_PLUGINLICENSE LICENSE_LGPL
#define PIL_PLUGINLICENSEURL URL_LGPL
#include <pils/plugin.h>
#endif
/*
* The SHA implementation is from mozilla.
* TODO: reuse the code from HBauth
*
* The Original Code is SHA 180-1 Reference Implementation
* (Compact version)
*
* The Initial Developer of the Original Code is Paul Kocher of
* Cryptography Research. Portions created by Paul Kocher are
* Copyright (C) 1995-9 by Cryptography Research, Inc. All
* Rights Reserved.
*
* Contributor(s):
*
* Paul Kocher
*/
typedef struct {
unsigned int H[5];
unsigned int W[80];
int lenW;
unsigned int sizeHi,sizeLo;
} SHA_CTX;
static SHA_CTX sha_ctx;
static void SHA1_Init(SHA_CTX *ctx);
static void SHA1_Update(SHA_CTX *ctx, const void *dataIn, int len);
static void SHA1_Final(unsigned char hashout[20], SHA_CTX *ctx);
static void shaHashBlock(SHA_CTX *ctx);
#ifdef DSKCM_STANDALONE
#define MAXMSG (256*1024)
#define PILCallLog(x, y, ...) fprintf(stderr, __VA_ARGS__)
#endif
#define DSKCM_ALIGN_SZ 512
#define DISKMETA_SZ 512
#define NODEMETA_SZ DSKCM_ALIGN_SZ
#define DSKCM_SLOT_SZ MAXMSG
#define DSKCM_MAX_NODES 64
#define DSKCM_SLOTS_PER_NODE 64
#define DSKCM_VERSION 1
#define DSKCM_ZEROOBJ(x) memset((&x), 0, sizeof(x))
struct diskmeta
{
int32_t version;
int32_t nodes;
int32_t slots;
} __attribute__((aligned(DSKCM_ALIGN_SZ)));
struct nodemeta
{
int32_t gen_number;
int32_t lws;
int32_t lrgn[DSKCM_MAX_NODES];
int32_t lrsn[DSKCM_MAX_NODES];
} __attribute__((aligned(DSKCM_ALIGN_SZ)));
struct slot
{
unsigned char chksum[20];
int32_t msg_size;
int32_t gen_number;
char msg[DSKCM_SLOT_SZ];
} __attribute__((aligned(DSKCM_ALIGN_SZ)));
#ifdef DSKCM_STANDALONE
static int init_dsk(int dskfd, int nodes, int slots);
static void usage (const char* fn);
#endif
static void print_chksum(const char* msg, unsigned char* r);
static int is_valid_dev(const char* dev);
static void mysleep(double tm);
static void compute_chksum(struct slot* p, unsigned char* r);
static int check_chksum(struct slot* p);
static int check_ns(int node, int slot);
static int dsk_nodemeta_only(int dskfd, int node);
static int dsk_slot_only(int dskfd, int node, int slot);
static int dsk_slot_get(int dskfd, int node, int slot);
static int dsk_slot_set(int dskfd, int node, int slot);
static ssize_t dsk_read(int dskfd, int node, int slot, int* gen_number, void** rbuf);
static int dsk_write(int dskfd, int slot, const char* msg, size_t msg_size);
static ssize_t read_new_arrival(int dskfd, int node, int* arri_node, void** rbuf);
//disk info
static int32_t dsk_max_nodes;
static int32_t dsk_slot_start;
static int32_t dsk_nodemeta_start;
static int32_t inited = 0;
static int32_t lrgn[DSKCM_MAX_NODES];
static int32_t lrsn[DSKCM_MAX_NODES];
static int32_t rns[DSKCM_MAX_NODES];
static int32_t toread = -1;
#ifndef DSKCM_STANDALONE
/* hua li de feng ge xian */
struct dskcm_private {
char* dskn;
int node;
int dskfd;
};
static struct hb_media* dskcm_new(const char* dev, int slot);
static int dskcm_parse(const char* line);
static int dskcm_open(struct hb_media* mp);
static int dskcm_close(struct hb_media* mp);
static void* dskcm_read(struct hb_media* mp, int *lenp);
static int dskcm_write(struct hb_media* mp, void* msg, int len);
static int dskcm_descr(char** buffer);
static int dskcm_mtype(char** buffer);
static int dskcm_isping(void);
static struct hb_media_fns dskcmOps ={
NULL, /* Create single object function */
dskcm_parse, /* whole-line parse function */
dskcm_open,
dskcm_close,
dskcm_read,
dskcm_write,
dskcm_mtype,
dskcm_descr,
dskcm_isping,
};
PIL_PLUGIN_BOILERPLATE2("1.0", Debug)
static const PILPluginImports* PluginImports;
static PILPlugin* OurPlugin;
static PILInterface* OurInterface;
static struct hb_media_imports* OurImports;
static void* interfprivate;
#define LOG PluginImports->log
#define MALLOC PluginImports->alloc
#define STRDUP PluginImports->mstrdup
#define FREE PluginImports->mfree
PIL_rc
PIL_PLUGIN_INIT(PILPlugin*us, const PILPluginImports* imports);
PIL_rc
PIL_PLUGIN_INIT(PILPlugin*us, const PILPluginImports* imports)
{
/* Force the compiler to do a little type checking */
(void)(PILPluginInitFun)PIL_PLUGIN_INIT;
PluginImports = imports;
OurPlugin = us;
/* Register ourself as a plugin */
imports->register_plugin(us, &OurPIExports);
/* Register our interface implementation */
return imports->register_interface(us, PIL_PLUGINTYPE_S
, PIL_PLUGIN_S
, &dskcmOps
, NULL /*close */
, &OurInterface
, (void*)&OurImports
, interfprivate);
}
/* XXX: why this ? */
static int
get_node(const char* node, int *s)
{
*s = (int)atoi(node);
return 0;
}
#define ISDSKCMOBJECT(mp) ((mp) && ((mp)->vf == (void*)&dskcmOps))
#define DSKCMASSERT(mp) g_assert(ISDSKCMOBJECT(mp))
static int
dskcm_mtype(char** buffer) {
*buffer = STRDUP(PIL_PLUGIN_S);
if (!*buffer) {
return 0;
}
return STRLEN_CONST(PIL_PLUGIN_S);
}
static int
dskcm_descr(char **buffer) {
const char constret[] = "Disk based broadcast";
*buffer = STRDUP(constret);
if (!*buffer) {
return 0;
}
return STRLEN_CONST(constret);
}
static int
dskcm_isping(void) {
return 0;
}
static int
dskcm_open(struct hb_media* mp)
{
struct dskcm_private * dpi;
DSKCMASSERT(mp);
dpi = (struct dskcm_private*) mp->pd;
if ((dpi->dskfd = open(dpi->dskn, O_RDWR | O_DIRECT)) < 0) {
return HA_FAIL;
}
PILCallLog(LOG, PIL_INFO
, "Disk based broadcasting started on device %s as node %d."
, dpi->dskn, dpi->node);
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG
, "dskcm_open : fd %d opened for reading/writing"
, dpi->dskfd);
}
return(HA_OK);
}
static int
dskcm_close(struct hb_media* mp)
{
struct dskcm_private * dpi;
int rc = HA_OK;
DSKCMASSERT(mp);
dpi = (struct dskcm_private *) mp->pd;
if (dpi->dskfd >= 0) {
if (close(dpi->dskfd) < 0) {
rc = HA_FAIL;
}
}
PILCallLog(LOG, PIL_INFO
, "Disk based broadcast heartbeat closed on device %s as node %d - Status: %d"
, dpi->dskn, dpi->node, rc);
return(rc);
}
void *
dskcm_read(struct hb_media* mp, int * lenp)
{
struct dskcm_private * dpi;
int numbytes;
int arri_node;
void* buf;
DSKCMASSERT(mp);
dpi = (struct dskcm_private *) mp->pd;
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG
, "dskcm_read : reading from device %s as node %d"
, dpi->dskn, dpi->node);
}
/* read available */
//read_new_arrival(int dskfd, int slot, int* arri_slot, void** rbuf);
numbytes = read_new_arrival(dpi->dskfd, dpi->node, &arri_node, &buf);
if (numbytes == -1) {
PILCallLog(LOG, PIL_CRIT
, "Error receiving from disk %s: %s"
, dpi->dskn, strerror(errno));
return NULL;
}
/* Avoid possible buffer overruns */
((char*)buf)[numbytes] = EOS;
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG, "got %d byte packet from %s as node %d"
, numbytes, dpi->dskn, dpi->node);
}
if (DEBUGPKTCONT && numbytes > 0) {
PILCallLog(LOG, PIL_DEBUG, "%s", (char*)buf);
}
*lenp = numbytes +1;
return buf;
}
static int
dskcm_write(struct hb_media* mp, void *pkt, int len)
{
struct dskcm_private * dpi;
int rc;
DSKCMASSERT(mp);
dpi = (struct dskcm_private *) mp->pd;
rc = dsk_write(dpi->dskfd, dpi->node, (const char*)pkt, len);
if (rc == -1) {
struct ha_msg* m;
int err = errno;
PILCallLog(LOG, PIL_CRIT, "Unable to send dskcm [%d] packet(len=%d): %s",
rc,len, strerror(err));
m = wirefmt2msg(pkt, len,MSG_NEEDAUTH);
if (m){
cl_log_message(LOG_ERR, m);
ha_msg_del(m);
}
errno = err;
return(HA_FAIL);
}
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG
, "dskcm_write : writing %d bytes to %s (node %d)"
, rc, dpi->dskn, dpi->node);
}
if (DEBUGPKTCONT) {
PILCallLog(LOG, PIL_DEBUG, "dskcm pkt out: [%s]", (char*)pkt);
}
mysleep(0.1);
return(HA_OK);
}
static int
dskcm_parse(const char* line)
{
const char* bp = line;
char dev[MAXLINE];
char token[MAXLINE];
int node;
int toklen;
struct hb_media *mp;
bp += strspn(bp, WHITESPACE);
toklen = strcspn(bp, WHITESPACE);
strncpy(dev, bp, toklen);
bp += toklen;
dev[toklen] = EOS;
if (*dev == EOS) {
PILCallLog(LOG, PIL_CRIT, "dskcm device is not set");
return HA_FAIL;
}
if (is_valid_dev(dev) == -1) {
PILCallLog(LOG, PIL_CRIT, "dskcm device [%s] is invalid or not set up properly", dev);
return HA_FAIL;
}
bp += strspn(bp, WHITESPACE);
toklen = strcspn(bp, WHITESPACE);
strncpy(token, bp, toklen);
bp += toklen;
token[toklen] = EOS;
if (*token == EOS) {
PILCallLog(LOG, PIL_CRIT, "dskcm [%s] missing slot", dev);
return HA_FAIL;
}
if (get_node(token, &node) < 0) {
PILCallLog(LOG, PIL_CRIT, "dskcm [%s] invalid node [%d]", dev, node);
return HA_FAIL;
}
if ((mp = dskcm_new(dev, node)) == NULL) {
return HA_FAIL;
}
OurImports->RegisterNewMedium(mp);
return HA_OK;
}
static struct dskcm_private*
new_dskcm_private(const char* dev, int node)
{
struct dskcm_private *dpi;
dpi = MALLOC(sizeof(struct dskcm_private));
if (dpi == NULL)
return NULL;
dpi->dskn = (char*)STRDUP(dev);
if (dpi->dskn == NULL) {
FREE(dpi);
return NULL;
}
dpi->node = node;
dpi->dskfd = -1;
return dpi;
}
static struct hb_media *
dskcm_new(const char * dev, int node)
{
struct dskcm_private* dpi;
struct hb_media * ret;
dpi = new_dskcm_private(dev, node);
if (dpi == NULL) {
PILCallLog(LOG, PIL_CRIT, "Create dskcm_private failed");
return(NULL);
}
ret = (struct hb_media*) MALLOC(sizeof(struct hb_media));
if (ret != NULL) {
char * name;
memset(ret, 0, sizeof(*ret));
ret->pd = (void*)dpi;
name = STRDUP(dev);
if (name != NULL) {
ret->name = name;
} else {
FREE(ret);
ret = NULL;
}
}
if (ret != NULL) {
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG,
"dskcm_new: returning ret (%s)",
ret->name);
}
}else{
FREE(dpi->dskn);
FREE(dpi);
if (DEBUGPKT) {
PILCallLog(LOG, PIL_DEBUG, "dskcm_new: ret was NULL");
}
}
return(ret);
}
#endif /* DSKCM_STANDALONE */
void mysleep(double tm)
{
double t = tm * 1000000;
if (random() % 2)
t += random() % 100000;
else
t -= random() % 100000;
usleep(t);
}
void
print_chksum(const char* msg, unsigned char* r)
{
int i;
printf("%s:\t", msg);
for (i = 0; i < 20; i++)
printf("%2x ", r[i]);
printf("\n");
}
void
compute_chksum(struct slot* p, unsigned char* r)
{
unsigned char hashout[20];
SHA1_Init(&sha_ctx);
SHA1_Update(&sha_ctx, p->msg, p->msg_size);
SHA1_Final(hashout, &sha_ctx);
memcpy(r, hashout, sizeof(unsigned char)*20);
}
int
check_chksum(struct slot* p)
{
unsigned char r[20];
compute_chksum(p, r);
if (memcmp(r, p->chksum, sizeof(r)) == 0)
return 0;
else
return -1;
}
static struct diskmeta diskmeta_i __attribute__((aligned(DSKCM_ALIGN_SZ)));
static struct nodemeta nodemeta_i __attribute__((aligned(DSKCM_ALIGN_SZ)));
//static struct nodemeta nodemeta_s[DSKCM_MAX_NODES] __attribute__((aligned(DSKCM_ALIGN_SZ)));
static struct slot readslot_i __attribute__((aligned(DSKCM_ALIGN_SZ)));
static struct slot writeslot_i __attribute__((aligned(DSKCM_ALIGN_SZ)));
int
is_valid_dev(const char* dev)
{
int dskfd = -1;
ssize_t rsz;
int i;
dskfd = open(dev, O_RDWR | O_DIRECT);
if (dskfd == -1) {
PILCallLog(LOG, PIL_CRIT,
"open(main): %s\n",
strerror(errno));
return -1;
}
DSKCM_ZEROOBJ(diskmeta_i);
rsz = pread(dskfd, &diskmeta_i, sizeof(struct diskmeta), 0);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread(init_dsk): %s\n",
strerror(errno));
close(dskfd);
return -1;
}
if (diskmeta_i.version != DSKCM_VERSION) {
PILCallLog(LOG, PIL_CRIT,
"This version of disk is not supported\n"
"Please re-initialize the disk.\n");
close(dskfd);
return -1;
}
i = diskmeta_i.nodes;
if (i <= 0 || i > DSKCM_MAX_NODES) {
PILCallLog(LOG, PIL_CRIT,
"Wrong number of slots.\n"
"Please re-initialize the disk.\n");
close(dskfd);
return -1;
}
dsk_max_nodes = i;
dsk_nodemeta_start = DISKMETA_SZ;
dsk_slot_start = DISKMETA_SZ + NODEMETA_SZ * dsk_max_nodes;
rsz = pread(dskfd, &readslot_i,
DSKCM_SLOT_SZ, dsk_slot_start+DSKCM_SLOT_SZ*(i-1));
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread last slot(is_valid_dev): %s\n",
strerror(errno));
close(dskfd);
return -1;
}
close(dskfd);
return 0;
}
int
check_ns(int node, int slot)
{
if (node < 0 || node >= diskmeta_i.nodes) {
PILCallLog(LOG, PIL_CRIT,
"check_ns: Not valid node number %d\n", node);
return -1;
}
if (slot < 0 || slot >= diskmeta_i.slots) {
PILCallLog(LOG, PIL_CRIT,
"check_ns: Not valid slot number %d\n", slot);
return -1;
}
return 0;
}
int
dsk_nodemeta_only(int dskfd, int node)
{
ssize_t rsz;
if (check_ns(node, 0) < 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_nodemeta_only: Not valid node or slot\n");
return -1;
}
rsz = pread(dskfd, &nodemeta_i, sizeof(struct nodemeta),
dsk_nodemeta_start + NODEMETA_SZ*node);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread(dsk_nodemeta_only): %s\n",
strerror(errno));
return -1;
}
return 0;
}
int
dsk_slot_only(int dskfd, int node, int slot)
{
ssize_t rsz;
if (check_ns(node, slot) < 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_slot_only: Not valid node or slot\n");
return -1;
}
rsz = pread(dskfd, &readslot_i, sizeof(struct slot),
dsk_slot_start + (DSKCM_SLOT_SZ*diskmeta_i.slots)*node + DSKCM_SLOT_SZ * slot);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread(dsk_slot_only): %s\n",
strerror(errno));
return -1;
}
return 0;
}
int
dsk_slot_get(int dskfd, int node, int slot)
{
ssize_t rsz;
if (check_ns(node, slot) < 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_slot_get: Not valid node or slot\n");
return -1;
}
rsz = pread(dskfd, &nodemeta_i, sizeof(struct nodemeta),
dsk_nodemeta_start + NODEMETA_SZ * node);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread(dsk_slot_get): %s\n",
strerror(errno));
return -1;
}
rsz = pread(dskfd, &readslot_i, sizeof(struct slot),
dsk_slot_start + (DSKCM_SLOT_SZ*diskmeta_i.slots)*node + DSKCM_SLOT_SZ * slot);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pread(dsk_slot_get): %s\n",
strerror(errno));
return -1;
}
return 0;
}
ssize_t
dsk_read(int dskfd, int node, int slot, int* gen_number, void** rbuf)
{
int ri;
printf("dsk_read: node %d, slot %d\n", node, slot);
ri = dsk_slot_get(dskfd, node, slot);
if (ri != 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_slot_get failed in dsk_read\n");
*rbuf = NULL;
return -1;
} // nodemeta_i & readslot_i are ready now.
while (check_chksum(&readslot_i) < 0) {
dsk_slot_only(dskfd, node, slot);
}
*gen_number = readslot_i.gen_number;
*rbuf = readslot_i.msg;
return readslot_i.msg_size;
}
ssize_t
read_new_arrival(int dskfd, int node, int* arri_node, void** rbuf)
{
int i = 0;
int ri;
int rg;
if (inited == 0 ) {
ri = dsk_slot_get(dskfd, node, 0);
if (ri != 0) {
PILCallLog(LOG, PIL_CRIT
, "Error get slot from disk: %s"
, strerror(errno));
return -1;
}
memcpy(lrgn, nodemeta_i.lrgn, sizeof(int32_t)*DSKCM_MAX_NODES);
memset(lrsn, 0, sizeof(int32_t)*DSKCM_MAX_NODES);
memset(rns, -1, sizeof(int32_t)*DSKCM_MAX_NODES);
inited = 1;
}
for(;;) {
if (toread >= 0)
break;
/*
printf("sizeof nodemeta_s is %d(%d)\n", sizeof(nodemeta_s), sizeof(nodemeta_s[0]));
printf("read %d bytes from %d\n",
(sizeof(struct nodemeta))*(diskmeta_i.nodes), dsk_nodemeta_start);
rsz = pread(dskfd, &nodemeta_s,
(sizeof(struct nodemeta))*(diskmeta_i.nodes),
dsk_nodemeta_start);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"Error pread: %s\n", strerror(errno));
return -1;
}
*/
for (i = 0; i < diskmeta_i.nodes; i++) {
if ( i == node ) continue;
ri = dsk_nodemeta_only(dskfd, i);
if (ri < 0) {
PILCallLog(LOG, PIL_CRIT, "Error get nodemeta\n");
return -1;
}
#if 0
printf("gen[%d]: %d, lrgn: %d\n", i, nodemeta_i.gen_number, lrgn[i]);
#endif
/*
if (nodemeta_s[i].gen_number > lrgn[i]) {
toread++;
rns[toread]=i;
}
*/
if (nodemeta_i.gen_number > lrgn[i]) {
toread++;
rns[toread] = i;
}
}
if (toread < 0) mysleep(0.1);
}
//nodemeta_i and readslot_i are renewed
i = rns[toread];
ri = dsk_read(dskfd, rns[toread], lrsn[i], &rg, rbuf);
if (ri < 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_read failed in read_new_available\n");
return -1;
}
i = rns[toread];
#if 1
if (rg == nodemeta_i.gen_number) { //happily, we finished reading this node
toread--;
lrsn[i] = (lrsn[i]+1)%(diskmeta_i.slots);
lrgn[i] = rg;
} else if (rg < nodemeta_i.gen_number) { //more to read in next iteration
lrsn[i] = (lrsn[i]+1)%(diskmeta_i.slots);
lrgn[i] = rg;
} else if (rg > nodemeta_i.gen_number) {
toread--;
}
#else
if (rg <= nodemeta_i.gen_number) {
toread--;
lrsn[i] = (lrsn[i]+1)%(diskmeta_i.slots);
lrgn[i] = rg;
} else if (rg > nodemeta_i.gen_number) {
toread--;
}
#endif
*rbuf = readslot_i.msg;
*arri_node = i;
return readslot_i.msg_size;
}
int
dsk_slot_set(int dskfd, int node, int slot)
{
ssize_t rsz;
if (check_ns(node, slot) < 0) {
PILCallLog(LOG, PIL_CRIT,
"Not valid node or slot number for writing\n");
return -1;
}
printf("slot write to offset %d\n", dsk_slot_start+(DSKCM_SLOT_SZ*diskmeta_i.slots)*node+DSKCM_SLOT_SZ*slot);
rsz = pwrite(dskfd, &writeslot_i, sizeof(struct slot),
dsk_slot_start+(DSKCM_SLOT_SZ*diskmeta_i.slots)*node+DSKCM_SLOT_SZ*slot);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pwrite(dsk_slot_set): %s\n", strerror(errno));
return -1;
}
printf("nodemeta write to offset %d. gen number is %d\n", dsk_nodemeta_start+NODEMETA_SZ*node, nodemeta_i.gen_number);
rsz = pwrite(dskfd, &nodemeta_i, sizeof(struct nodemeta),
dsk_nodemeta_start+NODEMETA_SZ*node);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pwrite nodemeta(dsk_slot_set): %s\n", strerror(errno));
return -1;
}
return 0;
}
int
dsk_write(int dskfd, int node, const char* msg, size_t msg_size)
{
int ws = 0;
int ri;
// mysleep(0.025);
ri = dsk_slot_get(dskfd, node, 0);
if (ri != 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_slot_get failed in dsk_write\n");
return -1;
}
if (msg_size > MAXMSG) {
PILCallLog(LOG, PIL_CRIT,
"message too big for dsk_write\n");
return -1;
}
writeslot_i.gen_number = nodemeta_i.gen_number + 1;
writeslot_i.msg_size = msg_size;
nodemeta_i.gen_number = nodemeta_i.gen_number + 1;
memcpy(writeslot_i.msg, msg, msg_size);
compute_chksum(&writeslot_i, writeslot_i.chksum);
ws = nodemeta_i.lws;
nodemeta_i.lws = (ws+1)%(diskmeta_i.slots);
#if 1
print_chksum("writeslot.chksum:", writeslot_i.chksum);
printf("write to node: %d, slot: %d\n", node, ws);
printf("msg: %s\n", writeslot_i.msg);
printf("msg_size: %d\n", writeslot_i.msg_size);
#endif
if (dsk_slot_set(dskfd, node, ws) < 0) {
PILCallLog(LOG, PIL_CRIT,
"dsk_slot_set failed in dsk_write\n");
return -1;
}
return 0;
}
static void SHA1_Init(SHA_CTX *ctx) {
int i;
ctx->lenW = 0;
ctx->sizeHi = ctx->sizeLo = 0;
/* Initialize H with the magic constants (see FIPS180 for constants)
*/
ctx->H[0] = 0x67452301;
ctx->H[1] = 0xefcdab89;
ctx->H[2] = 0x98badcfe;
ctx->H[3] = 0x10325476;
ctx->H[4] = 0xc3d2e1f0;
for (i = 0; i < 80; i++)
ctx->W[i] = 0;
}
static void SHA1_Update(SHA_CTX *ctx, const void *_dataIn, int len) {
const unsigned char *dataIn = _dataIn;
int i;
/* Read the data into W and process blocks as they get full
*/
for (i = 0; i < len; i++) {
ctx->W[ctx->lenW / 4] <<= 8;
ctx->W[ctx->lenW / 4] |= (unsigned int)dataIn[i];
if ((++ctx->lenW) % 64 == 0) {
shaHashBlock(ctx);
ctx->lenW = 0;
}
ctx->sizeLo += 8;
ctx->sizeHi += (ctx->sizeLo < 8);
}
}
static void SHA1_Final(unsigned char hashout[20], SHA_CTX *ctx) {
unsigned char pad0x80 = 0x80;
unsigned char pad0x00 = 0x00;
unsigned char padlen[8];
int i;
/* Pad with a binary 1 (e.g. 0x80), then zeroes, then length
*/
padlen[0] = (unsigned char)((ctx->sizeHi >> 24) & 255);
padlen[1] = (unsigned char)((ctx->sizeHi >> 16) & 255);
padlen[2] = (unsigned char)((ctx->sizeHi >> 8) & 255);
padlen[3] = (unsigned char)((ctx->sizeHi >> 0) & 255);
padlen[4] = (unsigned char)((ctx->sizeLo >> 24) & 255);
padlen[5] = (unsigned char)((ctx->sizeLo >> 16) & 255);
padlen[6] = (unsigned char)((ctx->sizeLo >> 8) & 255);
padlen[7] = (unsigned char)((ctx->sizeLo >> 0) & 255);
SHA1_Update(ctx, &pad0x80, 1);
while (ctx->lenW != 56)
SHA1_Update(ctx, &pad0x00, 1);
SHA1_Update(ctx, padlen, 8);
/* Output hash
*/
for (i = 0; i < 20; i++) {
hashout[i] = (unsigned char)(ctx->H[i / 4] >> 24);
ctx->H[i / 4] <<= 8;
}
/*
* Re-initialize the context (also zeroizes contents)
*/
SHA1_Init(ctx);
}
#define SHA_ROT(X,n) (((X) << (n)) | ((X) >> (32-(n))))
static void shaHashBlock(SHA_CTX *ctx) {
int t;
unsigned int A,B,C,D,E,TEMP;
for (t = 16; t <= 79; t++)
ctx->W[t] =
SHA_ROT(ctx->W[t-3] ^ ctx->W[t-8] ^ ctx->W[t-14] ^ ctx->W[t-16], 1);
A = ctx->H[0];
B = ctx->H[1];
C = ctx->H[2];
D = ctx->H[3];
E = ctx->H[4];
for (t = 0; t <= 19; t++) {
TEMP = SHA_ROT(A,5) + (((C^D)&B)^D) + E + ctx->W[t] + 0x5a827999;
E = D; D = C; C = SHA_ROT(B, 30); B = A; A = TEMP;
}
for (t = 20; t <= 39; t++) {
TEMP = SHA_ROT(A,5) + (B^C^D) + E + ctx->W[t] + 0x6ed9eba1;
E = D; D = C; C = SHA_ROT(B, 30); B = A; A = TEMP;
}
for (t = 40; t <= 59; t++) {
TEMP = SHA_ROT(A,5) + ((B&C)|(D&(B|C))) + E + ctx->W[t] + 0x8f1bbcdc;
E = D; D = C; C = SHA_ROT(B, 30); B = A; A = TEMP;
}
for (t = 60; t <= 79; t++) {
TEMP = SHA_ROT(A,5) + (B^C^D) + E + ctx->W[t] + 0xca62c1d6;
E = D; D = C; C = SHA_ROT(B, 30); B = A; A = TEMP;
}
ctx->H[0] += A;
ctx->H[1] += B;
ctx->H[2] += C;
ctx->H[3] += D;
ctx->H[4] += E;
}
#ifdef DSKCM_STANDALONE
int
main(int argc, char* argv[])
{
int initdsk_flag = 0;
int total_nodes = 0;
int mynode = -1;
int slots = DSKCM_SLOTS_PER_NODE;
int to_read = 0;
char *write_msg = NULL;
char *dskfn = NULL;
int dskfd;
int ri;
int c;
char *fn = argv[0];
opterr = 0;
while ((c = getopt(argc, argv, "i:n:s:d:rRw:")) != -1) {
switch(c) {
case 'i':
initdsk_flag = 1;
total_nodes = atoi(optarg);
break;
case 'n':
mynode = atoi(optarg);
break;
case 's':
slots = atoi(optarg);
break;
case 'd':
dskfn = optarg;
break;
case 'r':
to_read = 1;
break;
case 'R':
to_read = 2;
break;
case 'w':
write_msg = optarg;
break;
default:
usage(fn);
exit(1);
}
}
if (dskfn == NULL) {
printf("The disk name must be specified\n");
usage(fn);
exit(1);
}
srandom(time(NULL));
dskfd = open(dskfn, O_RDWR | O_DIRECT);
if (dskfd == -1)
{
perror("open(main)");
exit(1);
}
if (initdsk_flag == 1) {
if (total_nodes == 0) {
printf("The total number of slots must be specified\n");
usage(fn);
exit(1);
}
printf("Initializing %s...\n", dskfn);
init_dsk(dskfd, total_nodes, slots);
printf("Done\n");
exit(0);
}
if (is_valid_dev(dskfn) < 0) {
printf("invalid device\n");
usage(fn);
exit(1);
}
if (check_ns(mynode, 0) < 0) {
printf("The slot number is invalid\n");
usage(fn);
exit(1);
}
if (to_read == 2) {
char *msg = NULL;
ssize_t sz;
int rg;
sz = dsk_read(dskfd, mynode, slots, &rg, (void**)&msg);
if (sz > 0) {
printf("<%d>: %s\n", rg, msg);
}
exit(0);
}
if (to_read == 1) {
while (1) {
int arri_node;
size_t sz;
char *msg = NULL;
sz = read_new_arrival(dskfd, mynode, &arri_node, (void**)&msg);
if (sz>0) {
printf("<%2.0d>: %s\n", arri_node, msg);
}
}
}
if (write_msg)
dsk_write(dskfd, mynode, write_msg, strlen(write_msg)+1);
close(dskfd);
}
void
usage (const char* fn)
{
printf("usage: %s:\n", fn);
printf("\t -i <total slots>: number of slots in this disk.\n");
printf("\t -s <slot number>: the slot number this process uses\n");
printf("\t -d <disk>: the shared disk image to be used\n");
}
int
init_dsk(int dskfd, int nodes, int slots)
{
ssize_t rsz;
int ri;
int i, j;
if (nodes > DSKCM_MAX_NODES || nodes <= 0) {
PILCallLog(LOG, PIL_CRIT,
"The number of slots must between 0 and %d\n",
DSKCM_MAX_NODES);
return -1;
}
DSKCM_ZEROOBJ(diskmeta_i);
diskmeta_i.version = 1;
diskmeta_i.nodes = nodes;
diskmeta_i.slots = slots;
printf("Initialize device with %d nodes, and %d slots per node\n",
nodes, slots);
rsz = pwrite(dskfd, &diskmeta_i, sizeof(struct diskmeta), 0);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pwrite diskmeta(init_dsk)",
strerror(errno));
return -1;
}
printf("Initialize metadata for nodes\n");
DSKCM_ZEROOBJ(nodemeta_i);
printf("#");
for (i = 0; i < nodes; i++) {
rsz = pwrite(dskfd, &nodemeta_i,
NODEMETA_SZ,
DISKMETA_SZ+NODEMETA_SZ*i);
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT,
"pwrite nodemeta(init_dsk)",
strerror(errno));
return -1;
}
printf("#");
}
printf("\nNode meta initialize finished\n");
printf("Initialize slot\n");
DSKCM_ZEROOBJ(writeslot_i);
for (j = 0; j < nodes; j++) {
for (i = 0; i < slots; i++) {
rsz = pwrite(dskfd, &writeslot_i,
DSKCM_SLOT_SZ,
DISKMETA_SZ+NODEMETA_SZ*nodes+DSKCM_SLOT_SZ*(j*slots+i));
if (rsz == -1) {
PILCallLog(LOG, PIL_CRIT, "pwrite slot(init_dsk):", strerror(errno));
return -1;
}
printf("*(%d, %d)", j, i);
}
}
printf("\nSlot initialize finished\n");
return 0;
}
#endif /*DSKCM_STANDALONE*/
_______________________________________________________
Linux-HA-Dev: [email protected]
http://lists.linux-ha.org/mailman/listinfo/linux-ha-dev
Home Page: http://linux-ha.org/
