/*
* %CopyrightBegin%
*
* Copyright Ericsson AB 1998-2011. All Rights Reserved.
*
* The contents of this file are subject to the Erlang Public License,
* Version 1.1, (the "License"); you may not use this file except in
* compliance with the License. You should have received a copy of the
* Erlang Public License along with this software. If not, it can be
* retrieved online at http://www.erlang.org/.
*
* Software distributed under the License is distributed on an "AS IS"
* basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
* the License for the specific language governing rights and limitations
* under the License.
*
* %CopyrightEnd%
*
*/
#include <stdlib.h>
#include <string.h>
#include "eidef.h"
#include "eiext.h"
#include "reg.h"
#include "eisend.h"
#include "eirecv.h"
#include "ei_connect_int.h"
static int mn_start_restore(int fd, const erlang_pid *self, erlang_pid *mnesia, const char *mntab, long *count, long *maxkey,long *maxobj)
{
char buf[EISMALLBUF];
char *bufp=buf;
char tmpbuf[64];
int index = 0;
erlang_msg msg;
int arity;
int version;
int i;
int needlink;
int needmsg;
int msglen;
/* set up rpc arguments */
/* { PidFrom, { call, Mod, Fun, Args, user }} */
ei_encode_version(buf,&index);
ei_encode_tuple_header(buf,&index,2);
ei_encode_pid(buf,&index,self); /* PidFrom */
ei_encode_tuple_header(buf,&index,5);
ei_encode_atom(buf,&index,"call"); /* call */
ei_encode_atom(buf,&index,EI_MNESIA_MODULE); /* Mod */
ei_encode_atom(buf,&index,EI_MNESIA_RESTORE); /* Fun */
ei_encode_list_header(buf,&index,2); /* Args: [ table, self() ] */
ei_encode_atom(buf,&index,mntab);
ei_encode_pid(buf,&index,self);
ei_encode_empty_list(buf,&index);
ei_encode_atom(buf,&index,"user"); /* user */
/* make the rpc call */
if (ei_send_reg_encoded(fd,self,"rex",buf,index)) return -1;
/* get the reply: expect link and message (not sure which will come first though) */
needlink = needmsg = 1;
while (needlink || needmsg) {
/* get message */
index = EISMALLBUF;
while (!(i = ei_recv_internal(fd,&bufp,&index,&msg,&msglen,1,0))) index = EISMALLBUF;
switch (i) {
case ERL_LINK:
/* got link */
if (!needlink) return -1;
needlink = 0;
break;
case ERL_SEND:
/* got message - is it the right one? */
if (!needmsg) return -1;
else {
/* expecting { rex, { size, Pid, Count, MaxKey, MaxObj }} */
index = 0;
if (ei_decode_version(buf,&index,&version)
|| ei_decode_tuple_header(buf,&index,&arity)
|| (arity != 2)
|| ei_decode_atom(buf,&index,tmpbuf)
|| strcmp(tmpbuf,"rex")
|| ei_decode_tuple_header(buf,&index,&arity)
|| (arity != 5)
|| ei_decode_atom(buf,&index,tmpbuf)
|| strcmp(tmpbuf,EI_MNESIA_SIZE)
|| ei_decode_pid(buf,&index,mnesia)
|| ei_decode_long(buf,&index,count)
|| ei_decode_long(buf,&index,maxkey)
|| ei_decode_long(buf,&index,maxobj))
return -1; /* bad response from other side */
/* got msg */
needmsg = 0;
}
break;
default:
return -1; /* wasn't link or pid */
}
}
return 0;
}
static int mn_unlink(int fd)
{
erlang_msg msg;
char buf[EISMALLBUF];
char *bufp=buf;
int index;
int msglen;
/* wait for unlink or exit */
while (1) {
index = EISMALLBUF;
switch (ei_recv_internal(fd,&bufp,&index,&msg,&msglen,1,0)) {
case 0: continue;
case ERL_UNLINK: return 0;
default: return -1;
}
}
return 0;
}
/* decode an object and insert it into the table */
static int mn_decode_insert(ei_reg *reg, const char *msgbuf, int *index, char *key)
{
long keylen;
long objlen;
long objtype;
void *objbuf = NULL;
long i;
double f;
if (ei_decode_long(msgbuf,index,&keylen)
|| ei_decode_long(msgbuf,index,&objlen)
|| ei_decode_long(msgbuf,index,&objtype))
return -1;
/* decode key */
if (ei_decode_string(msgbuf,index,key)) {
if (objbuf) free(objbuf);
return -1;
}
/* finally! decode object and insert in table */
/* don't forget to fix attributes (dirty bit for example) */
/* FIXME: added cast but 64 bit trouble I think */
switch ((int)objtype & EI_REG_TYPEMASK) {
case EI_INT:
if (ei_decode_long(msgbuf,index,&i)) return -1;
ei_reg_setival(reg,key,i);
break;
case EI_FLT:
if (ei_decode_double(msgbuf,index,&f)) return -1;
ei_reg_setfval(reg,key,f);
break;
case EI_STR:
objbuf = NULL;
if (objlen > 0) {
if (!(objbuf = malloc(objlen))) return -1;
if (ei_decode_string(msgbuf,index,objbuf)) {
free(objbuf);
return -1;
}
ei_reg_setsval(reg,key,objbuf);
}
else {
/* just a pointer to nothing */
if (ei_decode_long(msgbuf,index,&i)) return -1;
ei_reg_setsval(reg,key,NULL);
}
break;
case EI_BIN:
objbuf = NULL;
if (objlen > 0) {
if (!(objbuf = malloc(objlen))) return -1;
if (ei_decode_binary(msgbuf,index,objbuf,&i)) {
free(objbuf);
return -1;
}
/* assert(i == objlen) */
ei_reg_setpval(reg,key,objbuf,objlen);
}
else {
/* just a pointer to nothing */
if (ei_decode_long(msgbuf,index,&i)) return -1;
ei_reg_setpval(reg,key,(void *)i,0);
}
break;
default:
/* unknown type */
if (objbuf) free(objbuf);
return -1;
} /* switch */
return 0;
}
/* help function passed to hash_foreach, to clear dirty bits */
/* use after successful restore */
static int clean_obj(const char *key, const void *p)
{
ei_reg_obj *obj = (ei_reg_obj *)p;
if (obj) obj->attr &= ~EI_DIRTY;
return 0;
}
int ei_reg_restore(int fd, ei_reg *reg, const char *mntab)
{
int i,j;
char tag[32];
char sbuf[EISMALLBUF];
char *dbuf = NULL;
char *msgbuf = NULL;
char *keybuf = NULL;
erlang_pid self;
erlang_pid mnesia = {"",0,0,0};
erlang_msg msg;
int index = 0;
int len = 0;
int msglen;
int version = 0;
int arity = 0;
long count = 0;
long maxkey = 0;
long maxobj = 0;
ei_cnode *ec;
if (!reg || !mntab) return -1; /* return EI_BADARG; */
/* make a self pid */
if ((ec = ei_fd_to_cnode(fd)) == NULL) {
return -1;
}
strcpy(self.node,ei_thisnodename(ec));
self.num = fd;
self.serial = 0;
self.creation = ei_thiscreation(ec);
if (mn_start_restore(fd,&self,&mnesia,mntab,&count,&maxkey,&maxobj)) {
/* send exit *only* if we have pid */
if (mnesia.node[0]) ei_send_exit(fd,&self,&mnesia,"bad response from rpc start");
return -1;
}
if (count <= 0) {
ei_send_exit(fd,&self,&mnesia,"nothing to do");
return 0;
}
/* make sure receive buffer can handle largest expected message */
len = maxkey + maxobj + 512;
if (len > EISMALLBUF)
if (!(dbuf = malloc(len))) {
ei_send_exit(fd,&self,&mnesia,"cannot allocate space for incoming data");
return -1;
}
msgbuf = (dbuf ? dbuf : sbuf);
/* allocate space for largest key */
if (!(keybuf = malloc(maxkey+1))) goto restore_failure;
/* get this ball rolling */
index = 0;
ei_encode_version(msgbuf,&index);
ei_encode_tuple_header(msgbuf,&index,2);
ei_encode_atom(msgbuf,&index,"send_records");
ei_encode_pid(msgbuf,&index,&self);
if (ei_send_encoded(fd,&mnesia,msgbuf,index)) goto restore_failure;
/* read as much as possible, until count or EXIT */
for (i=0; i<count; i++) {
index = len;
while ((j = ei_recv_internal(fd,&msgbuf,&index,&msg,&msglen,1,0)) == 0) index = len;
if (j<0) goto restore_failure;
/* decode the first part of the message */
index = 0;
if ((msg.msgtype != ERL_SEND)
|| ei_decode_version(msgbuf,&index,&version)
|| ei_decode_tuple_header(msgbuf,&index,&arity)
|| (arity != 6)
|| ei_decode_atom(msgbuf,&index,tag)
|| strcmp(tag,EI_MNESIA_RECV))
goto restore_failure;
/* decode the rest of the message and insert data into table */
if (mn_decode_insert(reg,msgbuf,&index,keybuf)) goto restore_failure;
}
if (keybuf) free(keybuf);
if (dbuf) free(dbuf);
/* wait for unlink */
if (mn_unlink(fd)) return -1;
/* clear all the dirty bits */
ei_hash_foreach(reg->tab,clean_obj);
/* success */
return 0;
restore_failure:
ei_send_exit(fd,&self,&mnesia,"restore failure");
if (keybuf) free(keybuf);
if (dbuf) free(dbuf);
return -1;
}
