/**********************************************************************
 *
 * Formatted to look for extended user location info
 *
 **********************************************************************/
static const char* userloc_format(
    const log4c_layout_t*       a_layout,
    const log4c_logging_event_t*a_event)
{
    static char buffer[4096];
    user_locinfo_t* uloc = NULL;

    sd_debug("Formatter s13_userloc checking location info for userdata %X",a_event->evt_loc->loc_data);
    if (a_event->evt_loc->loc_data != NULL)
    {
	sd_debug("Formatter s13_userloc getting a valid user location info pointer");
        uloc = (user_locinfo_t*) a_event->evt_loc->loc_data;
        sprintf(buffer, "[%s][HOST:%s][PID:%i][FILE:%s][LINE:%i][MSG:%s]",
		a_event->evt_category,  
		uloc->hostname, uloc->pid, a_event->evt_loc->loc_file,
		a_event->evt_loc->loc_line,a_event->evt_msg);

    }
    else
    {
        sprintf(buffer, "[%s]::[FILE:%s][LINE:%i][MSG::%s]", 
		a_event->evt_category,  
		a_event->evt_loc->loc_file,
		a_event->evt_loc->loc_line,a_event->evt_msg);
    }
    return buffer;
}

/*
 * Timeout after request is issued after 5s.
 *
 * Heart-beat message will be sent periodically with 1s interval.
 * If the node of the other end of fd fails, we'll detect it in 3s
 */
int set_keepalive(int fd)
{
	int val = 1;

	if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) < 0) {
		sd_debug("%m");
		return -1;
	}
	val = 5;
	if (setsockopt(fd, SOL_TCP, TCP_KEEPIDLE, &val, sizeof(val)) < 0) {
		sd_debug("%m");
		return -1;
	}
	val = 1;
	if (setsockopt(fd, SOL_TCP, TCP_KEEPINTVL, &val, sizeof(val)) < 0) {
		sd_debug("%m");
		return -1;
	}
	val = 3;
	if (setsockopt(fd, SOL_TCP, TCP_KEEPCNT, &val, sizeof(val)) < 0) {
		sd_debug("%m");
		return -1;
	}
	return 0;
}

/* Trim zero blocks of the beginning and end of the object. */
static int default_trim(int fd, uint64_t oid, const struct siocb *iocb,
			uint64_t *poffset, uint32_t *plen)
{
	trim_zero_blocks(iocb->buf, poffset, plen);

	if (iocb->offset < *poffset) {
		sd_debug("discard between %d, %ld, %" PRIx64, iocb->offset,
			 *poffset, oid);

		if (discard(fd, iocb->offset, *poffset) < 0)
			return -1;
	}

	if (*poffset + *plen < iocb->offset + iocb->length) {
		uint64_t end = iocb->offset + iocb->length;
		if (end == get_objsize(oid))
			/* This is necessary to punch the last block */
			end = round_up(end, BLOCK_SIZE);
		sd_debug("discard between %ld, %ld, %" PRIx64, *poffset + *plen,
			 end, oid);

		if (discard(fd, *poffset + *plen, end) < 0)
			return -1;
	}

	return 0;
}

static int err_to_sderr(char *path, uint64_t oid, int err)
{
	struct stat s;
	char *dir = dirname(path);

	sd_debug("%s", dir);
	switch (err) {
	case ENOENT:
		if (stat(dir, &s) < 0) {
			sd_err("%s corrupted", dir);
			return md_handle_eio(dir);
		}
		sd_debug("object %016" PRIx64 " not found locally", oid);
		return SD_RES_NO_OBJ;
	case ENOSPC:
		/* TODO: stop automatic recovery */
		sd_err("diskfull, oid=%"PRIx64, oid);
		return SD_RES_NO_SPACE;
	case EMFILE:
	case ENFILE:
	case EINTR:
	case EAGAIN:
	case EEXIST:
		sd_err("%m, oid=%"PRIx64, oid);
		/* make gateway try again */
		return SD_RES_NETWORK_ERROR;
	default:
		sd_err("oid=%"PRIx64", %m", oid);
		return md_handle_eio(dir);
	}
}

static int err_to_sderr(const char *path, uint64_t oid, int err)
{
	struct stat s;
	char p[PATH_MAX], *dir;

	/* Use a temporary buffer since dirname() may modify its argument. */
	pstrcpy(p, sizeof(p), path);
	dir = dirname(p);

	sd_debug("%s", path);
	switch (err) {
	case ENOENT:
		if (stat(dir, &s) < 0) {
			sd_err("%s corrupted", dir);
			return md_handle_eio(dir);
		}
		sd_debug("object %016" PRIx64 " not found locally", oid);
		return SD_RES_NO_OBJ;
	case ENOSPC:
		/* TODO: stop automatic recovery */
		sd_err("diskfull, oid=%"PRIx64, oid);
		return SD_RES_NO_SPACE;
	case EMFILE:
	case ENFILE:
	case EINTR:
	case EAGAIN:
	case EEXIST:
		sd_err("%m, oid=%"PRIx64, oid);
		/* make gateway try again */
		return SD_RES_NETWORK_ERROR;
	default:
		sd_err("oid=%"PRIx64", %m", oid);
		return md_handle_eio(dir);
	}
}

/*
 * If the node is gateway, this function only finds the store driver.
 * Otherwise, this function initializes the backend store
 */
int init_store_driver(bool is_gateway)
{
	char driver_name[STORE_LEN], *p;

	pstrcpy(driver_name, sizeof(driver_name), (char *)sys->cinfo.store);

	p = memchr(driver_name, '\0', STORE_LEN);
	if (!p) {
		/*
		 * If the driver name is not NUL terminated we are in deep
		 * trouble, let's get out here.
		 */
		sd_debug("store name not NUL terminated");
		return SD_RES_NO_STORE;
	}

	/*
	 * The store file might not exist in case this is a new sheep that
	 * never joined a cluster before.
	 */
	if (p == driver_name)
		return 0;

	sd_store = find_store_driver(driver_name);
	if (!sd_store) {
		sd_debug("store %s not found", driver_name);
		return SD_RES_NO_STORE;
	}

	if (is_gateway)
		return SD_RES_SUCCESS;

	return sd_store->init();
}

static int on_msg_error(struct xio_session *session,
			enum xio_status error,
			enum xio_msg_direction direction,
			struct xio_msg *msg,
			void *cb_user_context)
{
	/* struct server_data *sdata = (struct server_data *)cb_user_context; */

	if (direction == XIO_MSG_DIRECTION_OUT) {
		sd_debug("**** [%p] message %lu failed. reason: %s",
		       session, msg->sn, xio_strerror(error));
	} else {
		xio_release_response(msg);
		sd_debug("**** [%p] message %lu failed. reason: %s",
		       session, msg->request->sn, xio_strerror(error));
	}

	switch (error) {
	case XIO_E_MSG_FLUSHED:
		break;
	default:
		/* xio_disconnect(sdata->connection); */
		break;
	};

	return 0;
}

static long parse_byte_size (const char *astring)
{
    /* Parse size in bytes depending on the suffix.   Valid suffixes are KB, MB and GB */
    size_t sz = strlen (astring);
    long res = strtol(astring, (char **) NULL, 10);

    if (res <= 0)
	return 0;

    if (astring[ sz - 1 ] == 'B') {
	switch (astring[ sz - 2 ]) {
	    case 'K':
		res *= 1024;
		break;
	    case 'M':
		res *= 1024 * 1024;
		break;
	    case 'G':
		res *= 1024 * 1024 * 1024;
		break;
	    default:
		sd_debug("Wrong suffix parsing size in bytes for string %s, ignoring suffix", 
			astring);
	}
    }
    sd_debug("Parsed size parameter %s to value %ld",astring, res);
    return (res);
}

static int add_event(enum local_event_type type, struct local_node *lnode,
		      void *buf, size_t buf_len)
{
	struct local_node *n;
	struct local_event ev = {
		.type = type,
		.sender = *lnode,
	};

	ev.buf_len = buf_len;
	if (buf)
		memcpy(ev.buf, buf, buf_len);

	ev.nr_lnodes = get_nodes(ev.lnodes);

	switch (type) {
	case EVENT_JOIN:
		ev.lnodes[ev.nr_lnodes] = *lnode;
		ev.nr_lnodes++;
		break;
	case EVENT_LEAVE:
		xlremove(lnode, ev.lnodes, &ev.nr_lnodes, lnode_cmp);
		break;
	case EVENT_GATEWAY:
		n = xlfind(lnode, ev.lnodes, ev.nr_lnodes, lnode_cmp);
		n->gateway = true;
		break;
	case EVENT_NOTIFY:
	case EVENT_BLOCK:
		break;
	case EVENT_UPDATE_NODE:
		n = xlfind(lnode, ev.lnodes, ev.nr_lnodes, lnode_cmp);
		n->node = lnode->node;
		break;
	case EVENT_ACCEPT:
		abort();
	}

	sd_debug("type = %d, sender = %s", ev.type, lnode_to_str(&ev.sender));
	for (int i = 0; i < ev.nr_lnodes; i++)
		sd_debug("%d: %s", i, lnode_to_str(ev.lnodes + i));

	shm_queue_push(&ev);

	shm_queue_notify();

	return SD_RES_SUCCESS;
}

int default_create_and_write(uint64_t oid, const struct siocb *iocb)
{
	char path[PATH_MAX], tmp_path[PATH_MAX];
	int flags = prepare_iocb(oid, iocb, true);
	int ret, fd;
	uint32_t len = iocb->length;
	bool ec = is_erasure_obj(oid, iocb->copy_policy);
	size_t obj_size;

	sd_debug("%"PRIx64, oid);
	get_obj_path(oid, path, sizeof(path));
	get_tmp_obj_path(oid, tmp_path, sizeof(tmp_path));

	if (uatomic_is_true(&sys->use_journal) &&
	    journal_write_store(oid, iocb->buf, iocb->length,
				iocb->offset, true)
	    != SD_RES_SUCCESS) {
		sd_err("turn off journaling");
		uatomic_set_false(&sys->use_journal);
		flags |= O_DSYNC;
		sync();
	}

	fd = open(tmp_path, flags, sd_def_fmode);
	if (fd < 0) {
		if (errno == EEXIST) {
			/*
			 * This happens if node membership changes during object
			 * creation; while gateway retries a CREATE request,
			 * recovery process could also recover the object at the
			 * same time.  They should try to write the same date,
			 * so it is okay to simply return success here.
			 */
			sd_debug("%s exists", tmp_path);
			return SD_RES_SUCCESS;
		}

		sd_err("failed to open %s: %m", tmp_path);
		return err_to_sderr(path, oid, errno);
	}

	if (ec) {
		uint8_t policy = iocb->copy_policy ?:
			get_vdi_copy_policy(oid_to_vid(oid));
		int d;
		ec_policy_to_dp(policy, &d, NULL);
		obj_size = SD_DATA_OBJ_SIZE / d;
	} else

static inline void write_info_update(struct write_info *wi, int pos)
{
	sd_debug("%d, %d", wi->nr_sent, pos);
	wi->nr_sent--;
	memmove(wi->ent + pos, wi->ent + pos + 1,
		sizeof(struct write_info_entry) * (wi->nr_sent - pos));
}

void *mount3_mnt(struct svc_req *req, struct nfs_arg *arg)
{
	static mountres3 result;
	static int auth = AUTH_UNIX; /* FIXME: add auth support */
	static struct svc_fh fh;
	char *p = arg->mnt;
	uint32_t vid;
	int ret;

	sd_debug("%s", p);

	ret = sd_lookup_vdi(p, &vid);
	switch (ret) {
	case SD_RES_SUCCESS:
		fh.ino = fs_root_ino(vid);
		result.fhs_status = MNT3_OK;
		break;
	case SD_RES_NO_VDI:
		result.fhs_status = MNT3ERR_NOENT;
		goto out;
	default:
		result.fhs_status = MNT3ERR_SERVERFAULT;
		goto out;
	}

	result.mountres3_u.mountinfo.fhandle.fhandle3_len = sizeof(fh);
	result.mountres3_u.mountinfo.fhandle.fhandle3_val = (char *)&fh;
	result.mountres3_u.mountinfo.auth_flavors.auth_flavors_len = 1;
	result.mountres3_u.mountinfo.auth_flavors.auth_flavors_val = &auth;
out:
	return &result;
}

/* *********************************
   void serdisp_directgfx_init(dd)
   *********************************
   initialise
   *********************************
   dd     ... display descriptor
*/
void serdisp_directgfx_init(serdisp_t* dd) {
  char caption[40];
  SDL_Surface* screen;

  if (fp_SDL_Init(SDL_INIT_VIDEO) != 0) {
    sd_error(SERDISP_ENOTSUP, "%s(): unable to initialise SDL: %s", __func__, fp_SDL_GetError());
    sd_runtimeerror = 1;
    fp_SDL_Quit();
    return;
  }

  screen = serdisp_directgfx_internal_getStruct(dd)->screen = 
    (void*)fp_SDL_SetVideoMode(dd->width, dd->height, 0 /*sdl_depth*/, SDL_HWSURFACE | SDL_HWACCEL);

  if (screen == NULL) {
    sd_error(SERDISP_ENOTSUP, "%s(): unable to set video mode: %s", __func__, fp_SDL_GetError());
    sd_runtimeerror = 1;
    fp_SDL_Quit();
    return;
  }

  snprintf(caption, sizeof(caption)-1, "serdisplib::SDL %dx%dx%d", screen->w, screen->h, screen->format->BitsPerPixel);
  fp_SDL_WM_SetCaption(caption, NULL);

  sd_debug(2, "%s(): done with initialising", __func__);
}

void serdisp_t6963_checkready(serdisp_t* dd) {
  /* active-low signals are internally seen active-high because they will be auto-inverted later if needed */
  long td_clk1 =          SIG_CD         ;
  long td_clk2 = SIG_CE | SIG_CD | SIG_RD;
  long td_clk3 =          SIG_CD         ;

  int cnt = 0;

  /* check, if at least one of D0, D1, or D3 is high */
  long rc;     /* returned status byte */
  long rc_check = SIG_D0 | SIG_D1 | SIG_D3;

  if (dd->feature_backlight && dd->curr_backlight) {
    td_clk1 |= SIG_BACKLIGHT;
    td_clk2 |= SIG_BACKLIGHT;
    td_clk3 |= SIG_BACKLIGHT;
  }

  do {
    SDCONN_write(dd->sdcd, td_clk1, dd->sdcd->io_flags_writecmd);
    sdtools_nsleep( (dd->delay < 2) ? 2 : dd->delay);
    SDCONN_write(dd->sdcd, td_clk2, dd->sdcd->io_flags_writecmd);
    sdtools_nsleep( (dd->delay < 2) ? 2 : dd->delay);
    rc = SDCONN_read(dd->sdcd, dd->sdcd->io_flags_readstatus);
    sdtools_nsleep( (dd->delay < 2) ? 2 : dd->delay);
    SDCONN_write(dd->sdcd, td_clk3, dd->sdcd->io_flags_writecmd);
    sdtools_nsleep(dd->delay);
    cnt ++;
  } while (  (!(rc & rc_check)) && cnt < 10);

  if (cnt > 1)   
    sd_debug(1, "serdisp_t6963_checkready(): delay too small? (cnt = %d; rc = 0x%08lx)", cnt, rc);
}

static int gw_client_on_response(struct xio_session *session,
				 struct xio_msg *rsp,
				 int last_in_rxq,
				 void *cb_user_context)
{
	struct xio_forward_info_entry *fi_entry =
		(struct xio_forward_info_entry *)cb_user_context;
	struct xio_forward_info *fi = fi_entry->fi;

	struct xio_vmsg *pimsg = &rsp->in;
	struct xio_iovec_ex *isglist = vmsg_sglist(pimsg);

	int nents = vmsg_sglist_nents(pimsg), total = 0;

	sd_debug("response on fi_entry %p", fi_entry);

	for (int i = 0; i < nents; i++) {
		memcpy((char *)fi_entry->buf + total,
		       isglist[i].iov_base, isglist[i].iov_len);

		total += isglist[i].iov_len;
	}

	fi->nr_done++;
	if (fi->nr_done == fi->nr_send)
		xio_context_stop_loop(fi->ctx);

	return 0;
}

static int server_on_request(struct xio_session *session,
			     struct xio_msg *xio_req,
			     int last_in_rxq,
			     void *cb_user_conext)
{
	struct client_info *ci = (struct client_info *)cb_user_conext;
	struct sd_req *hdr;
	struct request *req;

	struct xio_iovec_ex *sglist = vmsg_sglist(&xio_req->in);
	int nents = vmsg_sglist_nents(&xio_req->in);

	sd_debug("on request: %p, %p, nents: %d", session, xio_req, nents);
	hdr = xio_req->in.header.iov_base;

	req = alloc_request(ci, hdr->data_length);
	memcpy(&req->rq, hdr, sizeof(req->rq));

	if (hdr->data_length && hdr->flags & SD_FLAG_CMD_WRITE) {
		sd_assert(nents == 1);
		req->data = sglist[0].iov_base;
	}

	xio_req->in.header.iov_base  = NULL;
	xio_req->in.header.iov_len  = 0;
	vmsg_sglist_set_nents(&xio_req->in, 0);

	ci->xio_req = xio_req;

	queue_request(req);

	xio_context_stop_loop(xio_get_main_ctx());
	return 0;
}

static int move_object_to_stale_dir(uint64_t oid, const char *wd,
				    uint32_t epoch, uint8_t ec_index,
				    struct vnode_info *vinfo, void *arg)
{
	char path[PATH_MAX], stale_path[PATH_MAX];
	uint32_t tgt_epoch = *(uint32_t *)arg;

	/* ec_index from md.c is reliable so we can directly use it */
	if (ec_index < SD_MAX_COPIES) {
		snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d",
			 md_get_object_dir(oid), oid, ec_index);
		snprintf(stale_path, PATH_MAX,
			 "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
			 md_get_object_dir(oid), oid, ec_index, tgt_epoch);
	} else {
		snprintf(path, PATH_MAX, "%s/%016" PRIx64,
			 md_get_object_dir(oid), oid);
		snprintf(stale_path, PATH_MAX, "%s/.stale/%016"PRIx64".%"PRIu32,
			 md_get_object_dir(oid), oid, tgt_epoch);
	}

	if (unlikely(rename(path, stale_path)) < 0) {
		sd_err("failed to move stale object %" PRIX64 " to %s, %m", oid,
		       path);
		return SD_RES_EIO;
	}

	sd_debug("moved object %"PRIx64, oid);
	return SD_RES_SUCCESS;
}

static int init_signal(void)
{
	sigset_t mask;
	int ret;

	sigemptyset(&mask);
	sigaddset(&mask, SIGTERM);
	sigprocmask(SIG_BLOCK, &mask, NULL);

	sigfd = signalfd(-1, &mask, SFD_NONBLOCK);
	if (sigfd < 0) {
		sd_err("failed to create a signal fd: %m");
		return -1;
	}

	ret = register_event(sigfd, signal_handler, NULL);
	if (ret) {
		sd_err("failed to register signal handler (%d)", ret);
		return -1;
	}

	sd_debug("register signal_handler for %d", sigfd);

	return 0;
}

int do_read(int sockfd, void *buf, int len, bool (*need_retry)(uint32_t epoch),
	    uint32_t epoch, uint32_t max_count)
{
	int ret, repeat = max_count;
reread:
	ret = read(sockfd, buf, len);
	if (ret == 0) {
		sd_debug("connection is closed (%d bytes left)", len);
		return 1;
	}
	if (ret < 0) {
		if (errno == EINTR)
			goto reread;
		/*
		 * Since we set timeout for read, we'll get EAGAIN even for
		 * blocking sockfd.
		 */
		if (errno == EAGAIN && repeat &&
		    (need_retry == NULL || need_retry(epoch))) {
			repeat--;
			goto reread;
		}

		sd_err("failed to read from socket: %d, %m", ret);
		return 1;
	}

	len -= ret;
	buf = (char *)buf + ret;
	if (len)
		goto reread;

	return 0;
}

int default_get_hash(uint64_t oid, uint32_t epoch, uint8_t *sha1)
{
	int ret;
	void *buf;
	struct siocb iocb = {};
	uint32_t length;
	bool is_readonly_obj = oid_is_readonly(oid);
	char path[PATH_MAX];

	ret = get_object_path(oid, epoch, path, sizeof(path));
	if (ret != SD_RES_SUCCESS)
		return ret;

	if (is_readonly_obj) {
		if (get_object_sha1(path, sha1) == 0) {
			sd_debug("use cached sha1 digest %s",
				 sha1_to_hex(sha1));
			return SD_RES_SUCCESS;
		}
	}

	length = get_store_objsize(oid);
	buf = valloc(length);
	if (buf == NULL)
		return SD_RES_NO_MEM;

	iocb.epoch = epoch;
	iocb.buf = buf;
	iocb.length = length;

	ret = default_read_from_path(oid, path, &iocb);
	if (ret != SD_RES_SUCCESS) {
		free(buf);
		return ret;
	}

	get_buffer_sha1(buf, length, sha1);
	free(buf);

	sd_debug("the message digest of %"PRIx64" at epoch %d is %s", oid,
		 epoch, sha1_to_hex(sha1));

	if (is_readonly_obj)
		set_object_sha1(path, sha1);

	return ret;
}