/* cf_svcmsocket_init
 * Initialize a multicast service/receive socket
 * Bind is done to INADDR_ANY - all interfaces
 *  */
int
cf_mcastsocket_init(cf_mcastsocket_cfg *ms)
{
    cf_socket_cfg *s = &(ms->s);

    if (0 > (s->sock = socket(AF_INET, SOCK_DGRAM, 0))) {
        cf_warning(CF_SOCKET, "multicast socket open error: %d %s", errno, cf_strerror(errno));
        return(-1);
    }

    cf_debug(CF_SOCKET, "mcast_socket init: socket %d",s->sock);

    // allows multiple readers on the same address
    uint yes=1;
    if (setsockopt(s->sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0) {
        cf_warning(CF_SOCKET, "multicast socket reuse failed: %d %s", errno, cf_strerror(errno));
        return(-1);
    }

    /* Set close-on-exec */
    fcntl(s->sock, F_SETFD, 1);

    // Bind to the incoming port on inaddr any
    memset(&s->saddr, 0, sizeof(s->saddr));
    s->saddr.sin_family = AF_INET;
    s->saddr.sin_addr.s_addr = INADDR_ANY;
    s->saddr.sin_port = htons(s->port);
    if (ms->tx_addr) {
        struct in_addr iface_in;
        memset((char *)&iface_in,0,sizeof(iface_in));
        iface_in.s_addr = inet_addr(ms->tx_addr);

        if(setsockopt(s->sock, IPPROTO_IP, IP_MULTICAST_IF, (const char*)&iface_in, sizeof(iface_in)) == -1) {
            cf_warning(CF_SOCKET, "IP_MULTICAST_IF: %d %s", errno, cf_strerror(errno));
            return(-1);
        }
    }
    unsigned char ttlvar = ms->mcast_ttl;
    if (ttlvar>0) {
        if (setsockopt(s->sock,IPPROTO_IP,IP_MULTICAST_TTL,(char *)&ttlvar,
                       sizeof(ttlvar)) == -1) {
            cf_warning(CF_SOCKET, "IP_MULTICAST_TTL: %d %s", errno, cf_strerror(errno));
        } else {
            cf_info(CF_SOCKET, "setting multicast TTL to be %d",ttlvar);
        }
    }
    while (0 > (bind(s->sock, (struct sockaddr *)&s->saddr, sizeof(struct sockaddr)))) {
        cf_info(CF_SOCKET, "multicast socket bind failed: %d %s", errno, cf_strerror(errno));
    }

    // Register for the multicast group
    inet_pton(AF_INET, s->addr, &ms->ireq.imr_multiaddr.s_addr);
    ms->ireq.imr_interface.s_addr = htonl(INADDR_ANY);
    if (ms->tx_addr) {
        ms->ireq.imr_interface.s_addr = inet_addr(ms->tx_addr);
    }
    setsockopt(s->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const void *)&ms->ireq, sizeof(struct ip_mreq));

    return(0);
}

//------------------------------------------------
// Dump a histogram to log.
//
// Note - DO NOT change the log output format in
// this method - tools such as as_log_latency
// assume this format.
//
// TODO - print the scale so as_log_latency can
// label its columns appropriately.
//
void
histogram_dump(histogram *h)
{
	int b;
	uint64_t counts[N_BUCKETS];

	for (b = 0; b < N_BUCKETS; b++) {
		counts[b] = cf_atomic64_get(h->counts[b]);
	}

	int i = N_BUCKETS;
	int j = 0;
	uint64_t total_count = 0;

	for (b = 0; b < N_BUCKETS; b++) {
		if (counts[b] != 0) {
			if (i > b) {
				i = b;
			}

			j = b;
			total_count += counts[b];
		}
	}

	char buf[100];
	int pos = 0;
	int k = 0;

	buf[0] = '\0';

	cf_info(AS_INFO, "histogram dump: %s (%zu total)", h->name, total_count);

	for ( ; i <= j; i++) {
		if (counts[i] == 0) { // print only non-zero columns
			continue;
		}

		int bytes = sprintf(buf + pos, " (%02d: %010zu) ", i, counts[i]);

		if (bytes <= 0) {
			cf_info(AS_INFO, "histogram dump error");
			return;
		}

		pos += bytes;

		if ((k & 3) == 3) { // maximum of 4 printed columns per log line
			 cf_info(AS_INFO, "%s", buf);
			 pos = 0;
			 buf[0] = '\0';
		}

		k++;
	}

	if (pos > 0) {
		cf_info(AS_INFO, "%s", buf);
	}
}

static int
is_connected(int fd)
{
	uint8_t buf[8];
	ssize_t rv = recv(fd, (void*)buf, sizeof(buf), MSG_PEEK | MSG_DONTWAIT | MSG_NOSIGNAL);
	
	if (rv == 0) {
		cf_debug("Connected check: Found disconnected fd %d", fd);
		return CONNECTED_NOT;
	}
	
	if (rv < 0) {
		if (errno == EBADF) {
			cf_warn("Connected check: Bad fd %d", fd);
			return CONNECTED_BADFD;
		}
		else if (errno == EWOULDBLOCK || errno == EAGAIN) {
			// The normal case.
			return CONNECTED;
		}
		else {
			cf_info("Connected check: fd %d error %d", fd, errno);
			return CONNECTED_ERROR;
		}
	}
	
	cf_info("Connected check: Peek got unexpected data for fd %d", fd);
	return CONNECTED;
}

void
as_bin_all_dump(as_storage_rd *rd, char *msg)
{
	cf_info(AS_BIN, "bin dump: %s: new nbins %d", msg, rd->n_bins);
	for (uint16_t i = 0; i < rd->n_bins; i++) {
		as_bin *b = &rd->bins[i];
		cf_info(AS_BIN, "bin %s: %d: bin %p inuse %d particle %p", msg, i, b, as_bin_inuse(b), b->particle);
	}
}

int
as_particle_get_flat_size(as_bin *b, size_t *flat_size)
{
	if (!b)
		return (-1);

	as_particle *p = as_bin_get_particle(b);
	uint8_t type = as_bin_get_particle_type(b);

	if (type == AS_PARTICLE_TYPE_NULL)
		return (-1);

#ifdef EXTRA_CHECKS
	// check the incoming type
	if (type < AS_PARTICLE_TYPE_NULL || type >= AS_PARTICLE_TYPE_MAX) {
		cf_info(AS_PARTICLE, "particle set: bad particle type %d, error", (int)type);
		return(-1);
	}
#endif

	uint32_t size = g_particle_get_flat_table[type](p);
	if (size == 0)	return(-1);
	*flat_size = size;
	return(0);
}

as_particle *as_particle_set_int(as_particle *p, as_particle_type type, void *data, uint32_t sz, bool data_in_memory)
{
	// convert the incoming buffer to a uint64_t

	uint64_t i;

	if (sz == 8) {
		i = __be64_to_cpup(data);
	}
	else if (sz == 4) {
		i = __be32_to_cpup(data);
	}
	else {
		i = int_convert(data, sz);
	}

	// The integer particle is never allocated anymore
	if (!p) {
		cf_info(AS_PARTICLE, "as_particle_set_int: null particle passed inf or integer. Error ");
		return (p);
	}
	as_particle_int *pi = (as_particle_int *)p;
	pi->i = i;
	return (p);
}

static uint32_t
udf_record_ttl(const as_rec * rec)
{
	int ret = udf_record_param_check(rec, UDF_BIN_NONAME, __FILE__, __LINE__);
	if (ret) {
		return 0;
	}

	udf_record * urecord = (udf_record *) as_rec_source(rec);


	if (urecord->flag & UDF_RECORD_FLAG_IS_SUBRECORD) {
		cf_debug(AS_UDF, "Return 0 TTL for subrecord ");
		return 0;
	}

	if ((urecord->flag & UDF_RECORD_FLAG_STORAGE_OPEN)) {
		uint32_t now = as_record_void_time_get();

		return urecord->r_ref->r->void_time > now ?
				urecord->r_ref->r->void_time - now : 0;
	}
	else {
		cf_info(AS_UDF, "Error in getting ttl: no record found");
		return 0; // since we can't indicate the record doesn't exist
	}
	return 0;
}

static int file_write(char * filename, uint8_t * content, size_t content_len, unsigned char * hash) {

	FILE *  file            = NULL;
	char    filepath[256]   = {0};

	file_resolve(filepath, filename, NULL);

	file = fopen(filepath, "w");
	if (file == NULL) {
		cf_warning(AS_UDF, "could not open udf put to %s: %s", filepath, cf_strerror(errno));
		return -1;
	}
	int r = fwrite(content, sizeof(char), content_len, file);
	if (r <= 0) {
		cf_info(AS_UDF, "could not write file %s %d", filepath, r);
		return -1;
	}

	fclose(file);
	file = NULL;

	file_generation(filepath, content, content_len, hash);

	return 0;
}

void
demarshal_file_handle_init()
{
	struct rlimit rl;

	pthread_mutex_lock(&g_file_handle_a_LOCK);

	if (g_file_handle_a == 0) {
		if (-1 == getrlimit(RLIMIT_NOFILE, &rl)) {
			cf_crash(AS_DEMARSHAL, "getrlimit: %s", cf_strerror(errno));
		}

		// Initialize the message pointer array and the unread byte counters.
		g_file_handle_a = cf_calloc(rl.rlim_cur, sizeof(as_proto *));
		cf_assert(g_file_handle_a, AS_DEMARSHAL, CF_CRITICAL, "allocation: %s", cf_strerror(errno));
		g_file_handle_a_sz = rl.rlim_cur;

		for (int i = 0; i < g_file_handle_a_sz; i++) {
			cf_queue_push(g_freeslot, &i);
		}

		pthread_create(&g_demarshal_reaper_th, 0, thr_demarshal_reaper_fn, 0);

		// If config value is 0, set a maximum proto size based on the RLIMIT.
		if (g_config.n_proto_fd_max == 0) {
			g_config.n_proto_fd_max = rl.rlim_cur / 2;
			cf_info(AS_DEMARSHAL, "setting default client file descriptors to %d", g_config.n_proto_fd_max);
		}
	}

	pthread_mutex_unlock(&g_file_handle_a_LOCK);
}

// Process a batch request.
static void
batch_process_request(batch_transaction* btr)
{
	// Keep the reaper at bay.
	btr->fd_h->last_used = cf_getms();

	cf_buf_builder* bb = 0;
	batch_build_response(btr, &bb);

	int fd = btr->fd_h->fd;

	if (bb) {
		int brv = batch_send_header(fd, bb->used_sz);

		if (brv == 0) {
			brv = batch_send(fd, bb->buf, bb->used_sz, MSG_NOSIGNAL | MSG_MORE);

			if (brv == 0) {
				brv = batch_send_final(fd, 0);
			}
		}
		cf_buf_builder_free(bb);
	}
	else {
		cf_info(AS_BATCH, " batch request: returned no local responses");
		batch_send_final(fd, 0);
	}

	batch_transaction_done(btr);
}

// This signal is our cue to roll the log.
void
as_sig_handle_hup(int sig_num)
{
    cf_info(AS_AS, "SIGHUP received, rolling log");

    cf_fault_sink_logroll();
}

int
as_msg_send_reply(as_file_handle *fd_h, uint32_t result_code, uint32_t generation,
		uint32_t void_time, as_msg_op **ops, as_bin **bins, uint16_t bin_count,
		as_namespace *ns, uint *written_sz, uint64_t trid, const char *setname)
{
	int rv = 0;

	// most cases are small messages - try to stack alloc if we can
	byte fb[MSG_STACK_BUFFER_SZ];
	size_t msg_sz = sizeof(fb);
//	memset(fb,0xff,msg_sz);  // helpful to see what you might not be setting

	uint8_t *msgp = (uint8_t *) as_msg_make_response_msg( result_code, generation,
					void_time, ops, bins, bin_count, ns,
					(cl_msg *)fb, &msg_sz, trid, setname);

	if (!msgp)	return(-1);

	if (fd_h->fd == 0) {
		cf_warning(AS_PROTO, "write to fd 0 internal error");
		cf_crash(AS_PROTO, "send reply: can't write to fd 0");
	}

//	cf_detail(AS_PROTO, "write fd %d",fd);

	size_t pos = 0;
	while (pos < msg_sz) {
		int rv = send(fd_h->fd, msgp + pos, msg_sz - pos, MSG_NOSIGNAL);
		if (rv > 0) {
			pos += rv;
		}
		else if (rv < 0) {
			if (errno != EWOULDBLOCK) {
				// common message when a client aborts
				cf_debug(AS_PROTO, "protocol write fail: fd %d sz %zd pos %zd rv %d errno %d", fd_h->fd, msg_sz, pos, rv, errno);
				as_end_of_transaction_force_close(fd_h);
				rv = -1;
				goto Exit;
			}
			usleep(1); // Yield
		} else {
			cf_info(AS_PROTO, "protocol write fail zero return: fd %d sz %d pos %d ", fd_h->fd, msg_sz, pos);
			as_end_of_transaction_force_close(fd_h);
			rv = -1;
			goto Exit;
		}
	}

	// good for stats as a higher layer
	if (written_sz) *written_sz = msg_sz;

	as_end_of_transaction_ok(fd_h);

Exit:
	if ((uint8_t *)msgp != fb)
		cf_free(msgp);

	return(rv);
}

/*
 * Reading local directory to get specific module item's contents.
 * In future if needed we can change this to reading from smd metadata. 
 */
int udf_cask_info_get(char *name, char * params, cf_dyn_buf * out) {

	int                 resp                = 0;
	char                filename[128]       = {0};
	int                 filename_len        = sizeof(filename);
	uint8_t *           content             = NULL;
	size_t              content_len         = 0;
	unsigned char       content_gen[256]    = {0};
	uint8_t             udf_type            = AS_UDF_TYPE_LUA;

	cf_debug(AS_INFO, "UDF CASK INFO GET");

	// get (required) script filename
	if ( as_info_parameter_get(params, "filename", filename, &filename_len) ) {
		cf_info(AS_INFO, "invalid or missing filename");
		cf_dyn_buf_append_string(out, "error=invalid_filename");
		return 0;
	}

	mod_lua_rdlock(&mod_lua);
	// read the script from filesystem
	resp = file_read(filename, &content, &content_len, content_gen);
	mod_lua_unlock(&mod_lua);
	if ( resp ) {
		switch ( resp ) {
			case 1 : {
				cf_dyn_buf_append_string(out, "error=not_found");
				break;
			}
			case 2 : {
				cf_dyn_buf_append_string(out, "error=empty");
				break;
			}
			default : {
				cf_dyn_buf_append_string(out, "error=unknown_error");
				break; // complier complains without a break;
			}
		}
	}
	else {
		// put back the result
		cf_dyn_buf_append_string(out, "gen=");
		cf_dyn_buf_append_string(out, (char *) content_gen);
		cf_dyn_buf_append_string(out, ";type=");
		cf_dyn_buf_append_string(out, as_udf_type_name[udf_type]);
		cf_dyn_buf_append_string(out, ";content=");
		cf_dyn_buf_append_buf(out, content, content_len);
		cf_dyn_buf_append_string(out, ";");
	}

	if ( content ) {
		cf_free(content);
		content = NULL;
	}

	return 0;
}

int udf_cask_info_remove(char *name, char * params, cf_dyn_buf * out) {

	char    filename[128]   = {0};
	int     filename_len    = sizeof(filename);
	char file_path[1024]	= {0};
	struct stat buf;

	cf_debug(AS_INFO, "UDF CASK INFO REMOVE");

	// get (required) script filename
	if ( as_info_parameter_get(params, "filename", filename, &filename_len) ) {
		cf_info(AS_UDF, "invalid or missing filename");
		cf_dyn_buf_append_string(out, "error=invalid_filename");
	}

	// now check if such a file-name exists :
	if (!g_config.mod_lua.user_path)
	{
		return -1;
	}

	snprintf(file_path, 1024, "%s/%s", g_config.mod_lua.user_path, filename);

	cf_debug(AS_INFO, " Lua file removal full-path is : %s \n", file_path);

	if (stat(file_path, &buf) != 0) {
		cf_info(AS_UDF, "failed to read file from : %s, error : %s", file_path, cf_strerror(errno));
		cf_dyn_buf_append_string(out, "error=file_not_found");
		return -1;
	}

	as_smd_delete_metadata(udf_smd_module_name, filename);

	// this is what an error would look like
	//    cf_dyn_buf_append_string(out, "error=");
	//    cf_dyn_buf_append_int(out, resp);

	cf_dyn_buf_append_string(out, "ok");

	cf_info(AS_UDF, "UDF module '%s' (%s) removed", filename, file_path);

	return 0;
}

// Initialize batch queues and worker threads.
int
as_batch_direct_init()
{
	uint32_t threads = g_config.n_batch_threads;
	cf_info(AS_BATCH, "Initialize batch-threads to %u", threads);
	int status = as_thread_pool_init_fixed(&batch_direct_thread_pool, threads, batch_worker, sizeof(batch_transaction), offsetof(batch_transaction,complete));
	
	if (status) {
		cf_warning(AS_BATCH, "Failed to initialize batch-threads to %u: %d", threads, status);
	}
	return status;
}

void
as_namespaces_init(bool cold_start_cmd, uint32_t instance)
{
	// Sanity-check the persistent memory scheme. TODO - compile-time assert.
	as_xmem_scheme_check();

	uint32_t stage_capacity = as_mem_check();

	if (cold_start_cmd) {
		cf_info(AS_NAMESPACE, "got cold-start command");
	}

	for (int i = 0; i < g_config.n_namespaces; i++) {
		as_namespace *ns = g_config.namespaces[i];

		// Cold start if manually forced.
		ns->cold_start = cold_start_cmd;

		as_namespace_setup(ns, instance, stage_capacity);

		// Done with temporary sets configuration array.
		if (ns->sets_cfg_array) {
			cf_free(ns->sets_cfg_array);
		}

		// set partition id inside partition object.
		for (uint i = 0; i < AS_PARTITIONS; i++) {
			ns->partitions[i].partition_id = i;
		}
		for (uint i = 0; i < AS_PARTITIONS; i++) {
			as_partition_init(&ns->partitions[i], ns, i);
		}

		as_sindex_init(ns);
	}

	// Register Secondary Index module with the majority merge policy callback.
	// Secondary index metadata is restored for all namespaces. Must be done
	// before as_storage_init() populates the indexes.
	int retval = as_smd_create_module(SINDEX_MODULE,
			as_smd_majority_consensus_merge, NULL, as_sindex_smd_accept_cb,
			NULL, as_sindex_smd_can_accept_cb, NULL);

	if (0 > retval) {
		cf_crash(AS_NAMESPACE, "failed to create SMD module \"%s\" (rv %d)",
				SINDEX_MODULE, retval);
	}

	// Wait for Secondary Index SMD to be completely restored.
	while (! g_sindex_smd_restored) {
		usleep(1000);
	}
}

int
as_particle_increment(as_bin *b, as_particle_type type, byte *buf, uint32_t sz, bool mc_compliant)
{
	if (type != AS_PARTICLE_TYPE_INTEGER) {
		cf_info(AS_PARTICLE, "attempt to add using non-integer");
		return(-1);
	}

	if (as_bin_is_integer(b)) {
		// standard case
		if (0 != as_particle_add_int(&b->iparticle, buf, sz, mc_compliant)) {
			return(-1);
		}
	}
	else {
		cf_info(AS_PARTICLE, "attempt to add to a non-integer");
		return(-2);
	}

	return( 0 );
}

as_set *as_namespace_init_set(as_namespace *ns, const char *set_name)
{
	if (! set_name) {
		return NULL;
	}

	uint32_t idx;
	cf_vmapx_err result = cf_vmapx_get_index(ns->p_sets_vmap, set_name, &idx);

	if (result == CF_VMAPX_ERR_NAME_NOT_FOUND) {
		as_set set;

		memset(&set, 0, sizeof(set));

		// Check name length just once, here at insertion. (Other vmap calls are
		// safe if name is too long - they return CF_VMAPX_ERR_NAME_NOT_FOUND.)
		strncpy(set.name, set_name, AS_SET_NAME_MAX_SIZE);

		if (set.name[AS_SET_NAME_MAX_SIZE - 1]) {
			set.name[AS_SET_NAME_MAX_SIZE - 1] = 0;

			cf_info(AS_NAMESPACE, "set name %s... too long", set.name);
			return NULL;
		}

		result = cf_vmapx_put_unique(ns->p_sets_vmap, &set, &idx);

		// Since this function can be called via info, need to handle race with
		// as_namespace_get_create_set() that returns CF_VMAPX_ERR_NAME_EXISTS.
		if (result != CF_VMAPX_OK && result != CF_VMAPX_ERR_NAME_EXISTS) {
			cf_warning(AS_NAMESPACE, "unexpected error %d", result);
			return NULL;
		}
	}
	else if (result != CF_VMAPX_OK) {
		// Should be impossible.
		cf_warning(AS_NAMESPACE, "unexpected error %d", result);
		return NULL;
	}

	as_set *p_set = NULL;

	if ((result = cf_vmapx_get_by_index(ns->p_sets_vmap, idx, (void**)&p_set)) != CF_VMAPX_OK) {
		// Should be impossible - just verified idx.
		cf_warning(AS_NAMESPACE, "unexpected error %d", result);
		return NULL;
	}

	return p_set;
}

// Initialize batch queues and worker threads.
void
as_batch_init()
{
	if (cf_atomic32_incr(&g_batch_init) != 1) {
		return;
	}

	cf_info(AS_BATCH, "Initialize %d batch worker threads.", g_config.n_batch_threads);
	g_batch_queue = cf_queue_create(sizeof(batch_transaction), true);
	int max = g_config.n_batch_threads;

	for (int i = 0; i < max; i++) {
		pthread_create(&g_batch_threads[i], 0, batch_process_queue, (void*)g_batch_queue);
	}
}

int
cf_ifaddr_get( cf_ifaddr **ifaddr, int *ifaddr_sz, uint8_t *buf, size_t bufsz)
{
    struct ifaddrs *ifa;
    int rv = getifaddrs(&ifa);
    if (rv != 0) {
        cf_info(CF_SOCKET, " could not get interface information: return value %d errno %d",rv,errno);
        return(-1);
    }
    struct ifaddrs *ifa_orig = ifa;

    // currently, return ipv4 only (?)
    int n_ifs = 0;
    while (ifa) {
        if ((ifa->ifa_addr) && (ifa->ifa_addr->sa_family == AF_INET)) {
            n_ifs++;
        }
        ifa = ifa->ifa_next;
    }

    if (bufsz < sizeof(cf_ifaddr) * n_ifs) {
        freeifaddrs(ifa_orig);
        return(-2);
    }

    *ifaddr_sz = n_ifs;
    *ifaddr = (cf_ifaddr *) buf;
    ifa = ifa_orig;
    int i = 0;
    while (ifa) {

        if ((ifa->ifa_addr) && (ifa->ifa_addr->sa_family == AF_INET))
        {

            (*ifaddr)[i].flags = ifa->ifa_flags;
            (*ifaddr)[i].family = ifa->ifa_addr->sa_family;
            memcpy( &((*ifaddr)[i].sa), ifa->ifa_addr, sizeof(struct sockaddr) );

            i++;
        }
        ifa = ifa->ifa_next;
    }

    freeifaddrs(ifa_orig);
    return(0);
}