text: /* -*- C -*- */
  /*
   * Copyright (c) 2018-2020 Seagate Technology LLC and/or its Affiliates
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   *
   * For any questions about this software or licensing,
   * please email opensource@seagate.com or cortx-questions@seagate.com.
   *
   * Original author:  Nachiket Sahasrabuddhe <nachiket.sahasrabuddhe@seagate.com>
   * Original creation date: 06-Sep-2018
   */
  
  #include <libgen.h>      /* dirname */
  
  #include "conf/schema.h" /* M0_CST_ISCS */
  #include "lib/memory.h"  /* m0_alloc m0_free */
  
  #include "c0appz.h"
  #include "c0appz_isc.h"  /* c0appz_isc_req */
  #include "libdemo.h"     /* mm_result */
  
  
  enum isc_comp_type {
  	ICT_PING,
  	ICT_MIN,
  	ICT_MAX,
  };
  
  /** Arguments for min/max operations */
  struct mm_args {
  	/** Length of an array. */
  	uint32_t  ma_len;
  	/** Array of doubles. */
  	double   *ma_arr;
  	/** Number of isc services. */
  	uint32_t  ma_svc_nr;
  	/** Length of a chunk per service. */
  	uint32_t  ma_chunk_len;
  	/** A service currently fed with input. */
  	uint32_t  ma_curr_svc_id;
  };
  
  static void fid_get(const char *f_name, struct m0_fid *fid)
  {
  	uint32_t f_key = m0_full_name_hash((const unsigned char*)f_name,
  					    strlen(f_name));
  	uint32_t f_cont = m0_full_name_hash((const unsigned char *)"libdemo",
  					    strlen("libdemo"));
  
  	m0_fid_set(fid, f_cont, f_key);
  }
  
  static int op_type_parse(const char *op_name)
  {
  	if (op_name == NULL)
  		return -EINVAL;
  	else if (!strcmp(op_name, "ping"))
  		return ICT_PING;
  	else if (!strcmp(op_name, "min"))
  		return ICT_MIN;
  	else if (!strcmp(op_name, "max"))
  		return ICT_MAX;
  	else
  		return -EINVAL;
  
  }
  
  static int file_to_array(const char *file_name, void **arr, uint32_t *arr_len)
  {
  	FILE     *fd;
  	uint32_t  i;
  	int       rc;
  	double   *val_arr;
  
  	fd = fopen(file_name, "r");
  	if (fd == NULL) {
  		fprintf(stderr, "error! Could not open file c0isc_data\n");
  		return -EINVAL;
  	}
  	fscanf(fd, "%d", arr_len);
  	/* XXX: Fix sizeof (double) with appropriate macro. */
  	M0_ALLOC_ARR(val_arr, *arr_len);
  	for (i = 0; i < *arr_len; ++i) {
  		rc = fscanf(fd, "%lf", &val_arr[i]);
  		if (rc == EOF) {
  			fprintf(stderr, "data file (%s) does not contain the "
  				"specified number of elements: %d\n",
  				file_name, *arr_len);
  			m0_free(val_arr);
  			fclose(fd);
  			return -EINVAL;
  		}
  	}
  	*arr = val_arr;
  	fclose(fd);
  	return 0;
  }
  
  static uint32_t isc_services_count(void)
  {
  	struct m0_fid start_fid = M0_FID0;
  	struct m0_fid proc_fid;
  	uint32_t      svc_nr = 0;
  	int           rc = 0;
  
  	while (rc == 0) {
  		rc = c0appz_isc_nxt_svc_get(&start_fid, &proc_fid,
  					    M0_CST_ISCS);
  		if (rc == 0)
  			++svc_nr;
  		start_fid = proc_fid;
  	}
  	return svc_nr;
  }
  
  static int op_init(enum isc_comp_type type, void **ip_args)
  {
  	struct mm_args *in_info;
  	double         *arr;
  	uint32_t        arr_len;
  	int             rc;
  
  	switch (type) {
  	case ICT_PING:
  		return 0;
  	case ICT_MIN:
  	case ICT_MAX:
  		M0_ALLOC_PTR(in_info);
  		if (in_info == NULL)
  			return -ENOMEM;
  		rc = file_to_array("c0isc_data", (void **)&arr, &arr_len);
  		if (rc != 0)
  			return rc;
  		in_info->ma_arr = arr;
  		in_info->ma_len = arr_len;
  		in_info->ma_curr_svc_id = 0;
  		in_info->ma_svc_nr = isc_services_count();
  		in_info->ma_chunk_len = arr_len / in_info->ma_svc_nr;
  		*ip_args = in_info;
  
  		return 0;
  	default:
  		fprintf(stderr, "Invalid operation %d\n", type);
  		return EINVAL;
  	}
  }
  
  static void op_fini(enum isc_comp_type op_type, struct mm_args *in_info,
  		    void *op_args)
  {
  	switch (op_type) {
  	case ICT_PING:
  		break;
  	case ICT_MIN:
  	case ICT_MAX:
  		if (in_info == NULL)
  			break;
  		m0_free(in_info->ma_arr);
  		m0_free(in_info);
  		m0_free(op_args);
  	}
  }
  
  static bool is_last_service(struct mm_args *in_info)
  {
  	return in_info->ma_curr_svc_id + 1 == in_info->ma_svc_nr;
  }
  
  static uint32_t offset_calc(struct mm_args *in_info)
  {
  	return in_info->ma_curr_svc_id * in_info->ma_chunk_len;
  }
  
  static uint32_t buf_len_calc(struct mm_args *in_info)
  {
  	/*
  	 * All services except one get ma_chunk_len of an array.
  	 * Need to incorporate the remainder into last service.
  	 */
  	return is_last_service(in_info) ?
  		in_info->ma_len - offset_calc(in_info) :
  		in_info->ma_chunk_len;
  }
  
  static int minmax_input_prepare(struct m0_buf *buf, struct m0_fid *comp_fid,
  				uint32_t *reply_len, enum isc_comp_type type,
  				struct mm_args *in_info)
  {
  	struct m0_buf buf_local = M0_BUF_INIT0;
  	uint32_t      buf_len;
  	uint32_t      offset;
  	int           rc;
  
  	*buf = M0_BUF_INIT0;
  	/** Calculate parameters relevant to array to be communicated. */
  	buf_len = buf_len_calc(in_info);
  	offset  = offset_calc(in_info);
  
  	/** A local buffer pointing to appropriate offset in the array. */
  	m0_buf_init(&buf_local, in_info->ma_arr + offset,
  		    buf_len * sizeof in_info->ma_arr[0]);
  	rc = m0_buf_copy_aligned(buf, &buf_local, M0_0VEC_SHIFT);
  	if (type == ICT_MIN)
  		fid_get("arr_min", comp_fid);
  	else
  		fid_get("arr_max", comp_fid);
  	*reply_len = CBL_DEFAULT_MAX;
  
  	return rc;
  }
  
  static int ping_input_prepare(struct m0_buf *buf, struct m0_fid *comp_fid,
  			      uint32_t *reply_len, enum isc_comp_type type)
  {
  	char *greeting;
  
  	*buf = M0_BUF_INIT0;
  	greeting = m0_strdup("Hello");
  	if (greeting == NULL) {
  		fprintf(stderr, "error! Out of memory for %d",
  			ICT_PING);
  		return -ENOMEM;
  	}
  	m0_buf_init(buf, greeting, strlen(greeting));
  	fid_get("hello_world", comp_fid);
  	*reply_len = CBL_DEFAULT_MAX;
  
  	return 0;
  }
  
  static int input_prepare(struct m0_buf *buf, struct m0_fid *comp_fid,
  			 uint32_t *reply_len, enum isc_comp_type type,
  			 void *ip_args)
  {
  	switch (type) {
  	case ICT_PING:
  		return ping_input_prepare(buf, comp_fid, reply_len, type);
  	case ICT_MIN:
  	case ICT_MAX:
  		return minmax_input_prepare(buf, comp_fid, reply_len, type,
  					    ip_args);
  	}
  	return -EINVAL;
  }
  
  struct mm_result *op_result(struct mm_result *x, struct mm_result *y,
  			    enum isc_comp_type op_type)
  {
  	switch (op_type) {
  	case ICT_MIN:
  		return x->mr_val <= y->mr_val ? x : y;
  	case ICT_MAX:
  		return x->mr_val >= y->mr_val ? x : y;
  	default:
  		return NULL;
  	}
  }
  
  static void *minmax_output_prepare(struct m0_buf *result,
  				   struct mm_args *in_info,
  				   struct mm_result *prev,
  				   enum isc_comp_type type)
  {
  	struct mm_result *new;
  
  	if (prev == NULL) {
  		M0_ALLOC_PTR(prev);
  		memcpy(prev, result->b_addr, sizeof prev[0]);
  		goto out;
  	}
  	new = result->b_addr;
  	/** Service sent index from sub-array. */
  	new->mr_idx += offset_calc(in_info);
  	/* Copy the current optimal value. */
  	*prev =  *op_result(prev, new, type);
   out:
  	/* Print the output. */
  	if (is_last_service(in_info))
  		fprintf(stderr, "idx=%d val=%lf\n", prev->mr_idx,
  			prev->mr_val);
  	/** Bookkeep the count of services communicated so far. */
  	++in_info->ma_curr_svc_id;
  	return prev;
  }
  
  /**
   * Apart from processing the output this can deserialize the buffer into
   * a structure relevant to the result of invoked computation.
   * and return the same.
   */
  static void* output_process(struct m0_buf *result, void *in_args,
  			    void *op_args, enum isc_comp_type type)
  {
  	switch (type) {
  	case ICT_PING:
  		printf ("Hello-%s@"FID_F"\n", (char *)result->b_addr,
  			FID_P((struct m0_fid *)op_args));
  		memset(result->b_addr, 'a', result->b_nob);
  		return NULL;
  	case ICT_MIN:
  	case ICT_MAX:
  		return minmax_output_prepare(result, in_args, op_args, type);
  	}
  	return NULL;
  }
  
  char *prog;
  
  static void usage_print()
  {
  	fprintf(stderr,"Usage:\n");
  	fprintf(stderr,"%s op_name\n", prog);
  	fprintf(stderr,"supported operations: ping, min, max\n");
  	fprintf(stderr, "See README");
  }
  
  
  int main(int argc, char **argv)
  {
  	struct m0_buf          buf;
  	struct m0_buf          result;
  	struct m0_fid          comp_fid;
  	struct c0appz_isc_req  req;
  	/* Holds arguments specific to an operation. */
  	void                  *ip_args;
  	/* Holds output specific to a computation. */
  	void                  *op_args;
  	struct m0_fid          svc_fid;
  	struct m0_fid          start_fid = M0_FID0;
  	uint32_t               reply_len;
  	int                    op_type;
  	int                    rc;
  
  	prog = basename(strdup(argv[0]));
  
  	/* check input */
  	if (argc != 2) {
  		usage_print();
  		return -1;
  	}
  
  	/* time in */
  	c0appz_timein();
  
  	c0appz_setrc(prog);
  	c0appz_putrc();
  
  	op_type = op_type_parse(argv[1]);
  	if (op_type == -EINVAL) {
  		usage_print(prog);
  		return -EINVAL;
  	}
  
  	m0trace_on = true;
  
  	/* initialize resources */
  	rc = c0appz_init(0);
  	if (rc != 0) {
  		fprintf(stderr,"error! c0appz_init() failed: %d\n", rc);
  		return -2;
  	}
  
  	if (isc_services_count() == 0) {
  		fprintf(stderr, "\nISC services are not started\n");
  		rc = -EINVAL;
  		goto out;
  	}
  
  	op_args = NULL;
  	ip_args = NULL;
  
  	/* Initialise the  parameters for operation. */
  	rc = op_init(op_type, &ip_args);
  	while (rc == 0) {
  		rc = c0appz_isc_nxt_svc_get(&start_fid, &svc_fid, M0_CST_ISCS);
  		if (rc != 0)
  			break;
  
  		/* Prepare arguments for computation. */
  		rc = input_prepare(&buf, &comp_fid, &reply_len,
  				   op_type, ip_args);
  		if (rc != 0) {
  			fprintf(stderr,
  				"error! Input preparation failed: %d\n", rc);
  			break;
  		}
  		rc = c0appz_isc_req_prepare(&req, &buf, &comp_fid, &result,
  					    &svc_fid, reply_len);
  		if (rc != 0) {
  			m0_buf_free(&buf);
  			fprintf(stderr,
  				"error! request preparation failed: %d\n", rc);
  			goto out;
  		}
  		/*
  		 * XXX: To achieve parallelism across services need to change
  		 * this to async.
  		 */
  		rc = c0appz_isc_req_send_sync(&req);
  		if (rc == 0) {
  			if (op_type == ICT_PING)
  				op_args = &svc_fid;
  			op_args = output_process(&result, ip_args, op_args,
  						 op_type);
  		} else {
  			fprintf(stderr, "Error from "FID_F" received: rc=%d\n"
  				"Check if the relevant library is loaded\n",
  				FID_P(&svc_fid), rc);
  		}
  		c0appz_isc_req_fini(&req);
  		start_fid = svc_fid;
  	}
  	/* Ignore the case when all services are iterated. */
  	if (rc == -ENOENT)
  		rc = 0;
  	op_fini(op_type, ip_args, op_args);
   out:
  	/* free resources*/
  	c0appz_free();
  
  	/* time out */
  	c0appz_timeout(0);
  
  	return rc;
  }
  
  /*
   *  Local variables:
   *  c-indentation-style: "K&R"
   *  c-basic-offset: 8
   *  tab-width: 8
   *  fill-column: 80
   *  scroll-step: 1
   *  End:
   */
  