Intermediate step - with 4K sectors, better threads handling (b4d4414a) · Commits · Elphel / elphel-apps-camogm

src/camogm.c

+38 −9

Original line number	Original line	Diff line number	Diff line
	@@ -163,6 +163,7 @@ static int get_disk_range(const char name, struct range rng);
	static int set_disk_range(const struct range *rng);		static int set_disk_range(const struct range *rng);
	static void get_disk_info(camogm_state *state);		static void get_disk_info(camogm_state *state);
	static struct timeval get_fpga_time(const int fd_fparsall, unsigned int port);		static struct timeval get_fpga_time(const int fd_fparsall, unsigned int port);

	int open_files(camogm_state *state);		int open_files(camogm_state *state);
	unsigned long getGPValue(unsigned int port, unsigned long GPNumber);		unsigned long getGPValue(unsigned int port, unsigned long GPNumber);
	void setGValue(unsigned int port, unsigned long GNumber, unsigned long value);		void setGValue(unsigned int port, unsigned long GNumber, unsigned long value);
	@@ -280,7 +281,12 @@ void camogm_init(camogm_state state, char pipe_name, uint16_t port_num)
	state->rawdev.mmap_default_size = MMAP_CHUNK_SIZE;		state->rawdev.mmap_default_size = MMAP_CHUNK_SIZE;
	state->sock_port = port_num;		state->sock_port = port_num;

	state->writer_params.data_ready = false;		// state->writer_params.data_ready = false;
			state->writer_params.chunk_page_prep = 0; // will prepare page 0
			state->writer_params.chunk_page_write = 0; // will write page 0



	state->writer_params.exit_thread = false;		state->writer_params.exit_thread = false;
	state->writer_params.state = STATE_STOPPED;		state->writer_params.state = STATE_STOPPED;

	@@ -692,7 +698,7 @@ int sendImageFrame(camogm_state *state)
	state->writer_params.segments = 1;		state->writer_params.segments = 1;
	}		}
	D3(fprintf(debug_file, "\tcirbuf_rp[%d] = 0x%x\t", port, state->cirbuf_rp[port]));		D3(fprintf(debug_file, "\tcirbuf_rp[%d] = 0x%x\t", port, state->cirbuf_rp[port]));
	D3(fprintf(debug_file, "_12_"));		D3(fprintf(debug_file, "_12_\n"));
	if (state->this_frame_params[port].color != COLORMODE_RAW) { // Tiff		if (state->this_frame_params[port].color != COLORMODE_RAW) { // Tiff
	state->packetchunks[state->chunk_index ].bytes = 2;		state->packetchunks[state->chunk_index ].bytes = 2;
	state->packetchunks[state->chunk_index++].chunk = (unsigned char*)trailer;		state->packetchunks[state->chunk_index++].chunk = (unsigned char*)trailer;
	@@ -701,6 +707,7 @@ int sendImageFrame(camogm_state *state)
	dbg_show_packetchunks(state);		dbg_show_packetchunks(state);


			D6(fprintf(debug_file, "_12.5_ before CAMOGM_FORMAT_XXX @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));

	switch (state->format) {		switch (state->format) {
	case CAMOGM_FORMAT_NONE: rslt = 0; break;		case CAMOGM_FORMAT_NONE: rslt = 0; break;
	@@ -741,14 +748,14 @@ int sendImageFrame(camogm_state *state)
	// end_time.tv_sec, end_time.tv_usec,		// end_time.tv_sec, end_time.tv_usec,
	// state->rawdev.last_jpeg_size));		// state->rawdev.last_jpeg_size));
	// D6(fprintf(debug_file, "Write speed: %d MB/s\n", mbps));		// D6(fprintf(debug_file, "Write speed: %d MB/s\n", mbps));
			D6(fprintf(debug_file, "Exit sendImageFrame() OK @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	return 0;		return 0;
	}		}

	static void dbg_show_packetchunks(camogm_state *state)		static void dbg_show_packetchunks(camogm_state *state)
	{		{
	int i;		int i;
	for (i = 0; i < FILE_CHUNKS_NUM; i++) {		for (i = 0; i < state->chunk_index; i++) {
	D4(fprintf(debug_file, "packetchunk[%d]: ptr = %p, size = %d\n", i, state->packetchunks[i].chunk, state->packetchunks[i].bytes));		D4(fprintf(debug_file, "packetchunk[%d]: ptr = %p, size = %d\n", i, state->packetchunks[i].chunk, state->packetchunks[i].bytes));
	}		}
	}		}
	@@ -1386,7 +1393,7 @@ int parse_cmd(camogm_state state, FILE npipe)
	//skip empty commands		//skip empty commands
	while (((cmd = getLineFromPipe(npipe))) && !cmd[0]) ;		while (((cmd = getLineFromPipe(npipe))) && !cmd[0]) ;
	if (!cmd) return 0; // nothing in the pipe		if (!cmd) return 0; // nothing in the pipe
	D2(fprintf(debug_file, "Got command: '%s'\n", cmd));		D2(fprintf(debug_file, "Got command: '%s' @ %06d -> ", cmd, get_fpga_usec(state->fd_fparmsall[0], 0)));

	// Acknowledge received command by copying frame number to per-daemon parameter		// Acknowledge received command by copying frame number to per-daemon parameter
	// GLOBALPARS(state->port_num, G_DAEMON_ERR + lastDaemonBit[state->port_num]) = GLOBALPARS(state->port_num, G_THIS_FRAME);		// GLOBALPARS(state->port_num, G_DAEMON_ERR + lastDaemonBit[state->port_num]) = GLOBALPARS(state->port_num, G_THIS_FRAME);
	@@ -1655,6 +1662,7 @@ int listener_loop(camogm_state *state)

	// enter main processing loop		// enter main processing loop
	while (process) {		while (process) {
			D6(fprintf(debug_file, "--------------- Listener loop @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	curr_port = select_port(state);		curr_port = select_port(state);
	state->port_num = curr_port;		state->port_num = curr_port;
	// look at command queue first		// look at command queue first
	@@ -1668,14 +1676,19 @@ int listener_loop(camogm_state *state)
	#endif		#endif
	cmd = parse_cmd(state, cmd_file);		cmd = parse_cmd(state, cmd_file);
	if (cmd) {		if (cmd) {
			D2(fprintf(debug_file, "DONE command @ %06d\n", get_fpga_usec(state->fd_fparmsall[0], 0)));

	if (cmd < 0) D0(fprintf(debug_file, "Unrecognized command\n"));		if (cmd < 0) D0(fprintf(debug_file, "Unrecognized command\n"));
	#ifdef USE_POLL		#ifdef USE_POLL
	}		}
	#endif		#endif
	} else if (state->prog_state == STATE_RUNNING) { // no commands in queue, started		continue;
			// } else if (state->prog_state == STATE_RUNNING) { // no commands in queue, started
			}
			if (state->prog_state == STATE_RUNNING) { // no commands in queue, started
	D6(fprintf(debug_file, "state->prog_state == STATE_RUNNING "));		D6(fprintf(debug_file, "state->prog_state == STATE_RUNNING "));
	rslt = -sendImageFrame(state);		rslt = -sendImageFrame(state);
	D6(fprintf(debug_file, " ==> %d",rslt));		D6(fprintf(debug_file, " ==> %d @ %06d\n",rslt,get_fpga_usec(state->fd_fparmsall[0], 0)));
	switch (rslt) {		switch (rslt) {
	case 0:		case 0:
	break; // frame sent OK, nothing to do (TODO: check file length/duration)		break; // frame sent OK, nothing to do (TODO: check file length/duration)
	@@ -1692,6 +1705,7 @@ int listener_loop(camogm_state *state)
	D0(fprintf(debug_file, "%s:line %d got broken frame (%d) while waiting for ready. Before that fp0=0x%x\n", __FILE__, __LINE__, fp1, fp0));		D0(fprintf(debug_file, "%s:line %d got broken frame (%d) while waiting for ready. Before that fp0=0x%x\n", __FILE__, __LINE__, fp1, fp0));
	rslt = CAMOGM_FRAME_BROKEN;		rslt = CAMOGM_FRAME_BROKEN;
	} else {		} else {
			D6(fprintf(debug_file, "No compressed frame ready, skipped 1 frame sync for port %d @ %06d\n",curr_port,get_fpga_usec(state->fd_fparmsall[0], 0)));
	break;		break;
	}		}
	}		}
	@@ -1760,10 +1774,14 @@ int listener_loop(camogm_state *state)
	// add port number to error code to facilitate debugging		// add port number to error code to facilitate debugging
	state->last_error_code = rslt + 100 * state->port_num;		state->last_error_code = rslt + 100 * state->port_num;
	} else if (state->prog_state == STATE_READING) {		} else if (state->prog_state == STATE_READING) {
			D6(fprintf(debug_file, "COMMAND_LOOP_DELAY start for STATE_READING@ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	usleep(COMMAND_LOOP_DELAY);		usleep(COMMAND_LOOP_DELAY);
			D6(fprintf(debug_file, "COMMAND_LOOP_DELAY end for STATE_READING@ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	} else { // not running, not starting		} else { // not running, not starting
	state->rawdev.thread_state = STATE_RUNNING;		state->rawdev.thread_state = STATE_RUNNING;
			D6(fprintf(debug_file, "COMMAND_LOOP_DELAY start for STATE_RUNNING@ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	usleep(COMMAND_LOOP_DELAY); // make it longer but interruptible by signals?		usleep(COMMAND_LOOP_DELAY); // make it longer but interruptible by signals?
			D6(fprintf(debug_file, "COMMAND_LOOP_DELAY end for STATE_RUNNING@ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	}		}
	#ifdef USE_POLL		#ifdef USE_POLL
	// } // if pfd.revents & POLLIN		// } // if pfd.revents & POLLIN
	@@ -2265,3 +2283,14 @@ struct timeval get_fpga_time(const int fd_fparsall, unsigned int port)

	return tv;		return tv;
	}		}

			/**
			* @brief Get current FPGA microseconds of time to use for timing of the code
			* @return FPGA current microseconds
			*/
			inline int get_fpga_usec(const int fd_fparsall, unsigned int port)
			{
			lseek(fd_fparsall, LSEEK_GET_FPGA_TIME, SEEK_END);
			// return (int) getGPValue(port, G_MICROSECONDS);
			return (int) GLOBALPARS(port, G_MICROSECONDS);
			}

src/camogm.h

+16 −6

Original line number	Original line	Diff line number	Diff line
	@@ -67,7 +67,9 @@
	#define COMMAND_LOOP_DELAY 500000		#define COMMAND_LOOP_DELAY 500000
	/** @brief File can be split up to this number of chunks */		/** @brief File can be split up to this number of chunks */
	#define FILE_CHUNKS_NUM 8		#define FILE_CHUNKS_NUM 8
			/** @brief Number of chunk pages to overlap preparation with SSD raw write */
			#define FILE_CHUNKS_PAGES 2
			#define FILE_CHUNKS_TOTAL ((FILE_CHUNKS_NUM) * (FILE_CHUNKS_PAGES))
	/**		/**
	* @enum state_flags		* @enum state_flags
	* @brief Program state flags		* @brief Program state flags
	@@ -145,25 +147,32 @@ struct writer_params {
	pthread_mutex_t writer_mutex; ///< synchronization mutex for main and writing threads		pthread_mutex_t writer_mutex; ///< synchronization mutex for main and writing threads
	pthread_cond_t writer_cond; ///< conditional variable indicating that writer thread can proceed with new frame		pthread_cond_t writer_cond; ///< conditional variable indicating that writer thread can proceed with new frame
	pthread_cond_t main_cond; ///< conditional variable indicating that main thread can update write pointers		pthread_cond_t main_cond; ///< conditional variable indicating that main thread can update write pointers
	bool data_ready; ///< flag indicating that new frame is ready for recording, access to this flag		// bool data_ready; ///< flag indicating that new frame is ready for recording, access to this flag
	///< must be protected with #writer_mutex. Set this flag in main thread and reset in		// ///< must be protected with #writer_mutex. Set this flag in main thread and reset in
	///< disk writing thread.		// ///< disk writing thread.
			// ///< For overlapping disk raw writes with chunks preparation (should be atomic - use #writer_mutex)
			int chunk_page_prep; ///< page of chunks being prepared. Incremented (mod) after data is prepared for raw write
			int chunk_page_write; ///< page of chunks to write. Incremented (mod) after recording to disk
	int last_ret_val; ///< error value return during last frame recording (if any occurred)		int last_ret_val; ///< error value return during last frame recording (if any occurred)
	bool exit_thread; ///< flag indicating that the writing thread should terminate		bool exit_thread; ///< flag indicating that the writing thread should terminate
	int state; ///< the state of disk writing thread		int state; ///< the state of disk writing thread
	int segments; ///< the number of segments in frame		int segments; ///< the number of segments in frame

	struct iovec *data_chunks; ///< a set of vectors pointing to aligned frame data buffers		struct iovec *data_chunks; ///< a set of vectors pointing to aligned frame data buffers
			struct iovec was_CHUNK_REM; ///< extracting to a separate entity data_chunks[CHUNK_REM]
			// struct iovec *was_CHUNK_COMMON; ///< extracting to a separate entity data_chunks[CHUNK_COMMON]
	struct iovec prev_rem_vect; ///< vector pointing to the remainder of the previous frame		struct iovec prev_rem_vect; ///< vector pointing to the remainder of the previous frame
	unsigned char *rem_buff; ///< buffer containing the unaligned remainder of the current frame		unsigned char *rem_buff; ///< buffer containing the unaligned remainder of the current frame
	unsigned char *prev_rem_buff; ///< buffer containing the unaligned remainder of the previous frame		unsigned char *prev_rem_buff; ///< buffer containing the unaligned remainder of the previous frame
	unsigned char *common_buff; ///< buffer for aligned JPEG header		// get rid of common_buff
			// unsigned char *common_buff; ///< buffer for aligned JPEG header // make multiple?
			unsigned char *common_buffs[FILE_CHUNKS_PAGES]; ///< buffer for aligned JPEG header // make multiple?
	uint64_t lba_start; ///< disk starting LBA		uint64_t lba_start; ///< disk starting LBA
	uint64_t lba_current; ///< current write position in LBAs		uint64_t lba_current; ///< current write position in LBAs
	uint64_t lba_end; ///< disk last LBA		uint64_t lba_end; ///< disk last LBA

	time_t stat_update; ///< time when status file was updated		time_t stat_update; ///< time when status file was updated
	bool dummy_read; ///< inable dummy read cycle (debug feature)		bool dummy_read; ///< enable dummy read cycle (debug feature)
	};		};
	/**		/**
	* @struct camogm_state		* @struct camogm_state
	@@ -272,5 +281,6 @@ void setGValue(unsigned int port, unsigned long GNumber, unsigned long value);
	int waitDaemonEnabled(unsigned int port, int daemonBit);		int waitDaemonEnabled(unsigned int port, int daemonBit);
	int isDaemonEnabled(unsigned int port, int daemonBit);		int isDaemonEnabled(unsigned int port, int daemonBit);
	int is_fd_valid(int fd);		int is_fd_valid(int fd);
			int get_fpga_usec(const int fd_fparsall, unsigned int port);

	#endif /* _CAMOGM_H */		#endif /* _CAMOGM_H */

src/camogm_align.c

+280 −185

File changed.

Preview size limit exceeded, changes collapsed.

src/camogm_align.h

+7 −6

Original line number	Original line	Diff line number	Diff line
	@@ -22,8 +22,9 @@
	#include <sys/types.h>		#include <sys/types.h>

	#include "camogm.h"		#include "camogm.h"
			// TODO: Change PHY_BLOCK_SIZE to 4096 as it is 4096!
	#define PHY_BLOCK_SIZE 512 ///< Physical disk block size		//#define PHY_BLOCK_SIZE 512 ///< Physical disk block size
			#define PHY_BLOCK_SIZE 4096 ///< Physical disk block size
	#define JPEG_MARKER_LEN 2 ///< The size in bytes of JPEG marker		#define JPEG_MARKER_LEN 2 ///< The size in bytes of JPEG marker
	#define JPEG_SIZE_LEN 2 ///< The size in bytes of JPEG marker length field		#define JPEG_SIZE_LEN 2 ///< The size in bytes of JPEG marker length field
	#define INCLUDE_REM 1 ///< Include REM buffer to total size calculation		#define INCLUDE_REM 1 ///< Include REM buffer to total size calculation
	@@ -73,10 +74,10 @@ enum {
	int init_align_buffers(camogm_state *state);		int init_align_buffers(camogm_state *state);
	void deinit_align_buffers(camogm_state *state);		void deinit_align_buffers(camogm_state *state);
	void align_frame(camogm_state *state);		void align_frame(camogm_state *state);
	void reset_chunks(struct iovec *vects, int all);		void reset_chunks(camogm_state *state, int all, int page);
	int update_lba(camogm_state *state);		//int update_lba(camogm_state *state);
	int get_data_buffers(camogm_state state, struct iovec mapped, size_t all_sz);		int get_data_buffers(camogm_state state, struct iovec mapped, size_t mapped_sz, int page);
	int prep_last_block(camogm_state *state);		int prep_last_block(camogm_state *state, int page);
	off64_t lba_to_offset(uint64_t lba);		off64_t lba_to_offset(uint64_t lba);

src/camogm_jpeg.c

+156 −36

Original line number	Original line	Diff line number	Diff line
	@@ -35,6 +35,7 @@
	#include "camogm_read.h"		#include "camogm_read.h"
	#include "camogm_align.h"		#include "camogm_align.h"


	/** State file record format. It includes device path in /dev, starting, current and ending LBAs */		/** State file record format. It includes device path in /dev, starting, current and ending LBAs */
	#define STATE_FILE_FORMAT "%s\t%llu\t%llu\t%llu\n"		#define STATE_FILE_FORMAT "%s\t%llu\t%llu\t%llu\n"
	/** Status file update period, in seconds */		/** Status file update period, in seconds */
	@@ -322,35 +323,43 @@ int camogm_frame_jpeg(camogm_state *state)
	for (int i = 0; i < state->chunk_index - 1; i++) {		for (int i = 0; i < state->chunk_index - 1; i++) {
	D6(fprintf(debug_file, "ptr: %p, length: %ld\n", state->packetchunks[i + 1].chunk, state->packetchunks[i + 1].bytes));		D6(fprintf(debug_file, "ptr: %p, length: %ld\n", state->packetchunks[i + 1].chunk, state->packetchunks[i + 1].bytes));
	}		}
			D6(fprintf(debug_file, "waiting for empty buffer @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	// next frame is ready for recording, signal this to the writer thread		// next frame is ready for recording, signal this to the writer thread
	pthread_mutex_lock(&state->writer_params.writer_mutex);		pthread_mutex_lock(&state->writer_params.writer_mutex);
	while (state->writer_params.data_ready)		// while (state->writer_params.data_ready)
			// pthread_cond_wait(&state->writer_params.main_cond, &state->writer_params.writer_mutex);
			// do not run ahead by >= FILE_CHUNKS_PAGES (0 or 1 for just FILE_CHUNKS_PAGES == 2)
			while ((state->writer_params.chunk_page_prep - state->writer_params.chunk_page_write + (2 * FILE_CHUNKS_PAGES)) % (2 * FILE_CHUNKS_PAGES) >= FILE_CHUNKS_PAGES)
	pthread_cond_wait(&state->writer_params.main_cond, &state->writer_params.writer_mutex);		pthread_cond_wait(&state->writer_params.main_cond, &state->writer_params.writer_mutex);
	D6(fprintf(debug_file, "_13a_"));		pthread_mutex_unlock(&state->writer_params.writer_mutex);
			D6(fprintf(debug_file, "_13a_: got empty buffer @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
	D6(fprintf(debug_file, "\n"));
	align_frame(state);		align_frame(state);
			/* // Moved to write thread
	if (update_lba(state) == 1) {		if (update_lba(state) == 1) {
	D0(fprintf(debug_file, "The end of block device reached, continue recording from start\n"));		D0(fprintf(debug_file, "The end of block device reached, continue recording from start\n"));
	lseek(state->writer_params.blockdev_fd, 0, SEEK_SET);		lseek(state->writer_params.blockdev_fd, 0, SEEK_SET);
	}		}
	D6(fprintf(debug_file, "Block device positions: start = %llu, current = %llu, end = %llu\n",		D6(fprintf(debug_file, "Block device positions: start = %llu, current = %llu, end = %llu\n",
	state->writer_params.lba_start, state->writer_params.lba_current, state->writer_params.lba_end));		state->writer_params.lba_start, state->writer_params.lba_current, state->writer_params.lba_end));
			*/
	// proceed if last frame was recorded without errors		// proceed if last frame was recorded without errors
			D6(fprintf(debug_file, "_13a1_:frame aligned @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
			pthread_mutex_lock(&state->writer_params.writer_mutex);
	if (state->writer_params.last_ret_val == 0) {		if (state->writer_params.last_ret_val == 0) {
	state->writer_params.data_ready = true;		state->writer_params.chunk_page_prep = (state->writer_params.chunk_page_prep + 1) % (2 * FILE_CHUNKS_PAGES);
			// state->writer_params.data_ready = true;
	pthread_cond_signal(&state->writer_params.writer_cond);		pthread_cond_signal(&state->writer_params.writer_cond);
			D6(fprintf(debug_file, "_13b_: chunk_page_prep = %d (%d), signal writer_cond @ %06d\n",state->writer_params.chunk_page_prep, state->writer_params.chunk_page_write, get_fpga_usec(state->fd_fparmsall[0], 0)));
	}		}
	pthread_mutex_unlock(&state->writer_params.writer_mutex);		pthread_mutex_unlock(&state->writer_params.writer_mutex);
	if (state->writer_params.last_ret_val != 0) {		if (state->writer_params.last_ret_val != 0) {
	return state->writer_params.last_ret_val;		return state->writer_params.last_ret_val;
	}		}

	// update status file if time has come		// update status file if time has come
	curr_time = time(NULL);		curr_time = time(NULL);
	if (difftime(curr_time, state->writer_params.stat_update) > STAT_UPDATE_PERIOD) {		if (difftime(curr_time, state->writer_params.stat_update) > STAT_UPDATE_PERIOD) {
	save_state_file(state);		save_state_file(state);
			D6(fprintf(debug_file, "_13c_: saved state file @ %06d\n", get_fpga_usec(state->fd_fparmsall[0], 0)));
	}		}
	}		}

	@@ -364,7 +373,7 @@ int camogm_frame_jpeg(camogm_state *state)
	* @param state a pointer to a structure containing current state		* @param state a pointer to a structure containing current state
	* @return 0 if the device was closed successfully and -1 otherwise		* @return 0 if the device was closed successfully and -1 otherwise
	*/		*/
	int camogm_end_jpeg(camogm_state *state)		int camogm_end_jpeg(camogm_state *state) // TODO: Resolve multippage !
	{		{
	int ret = 0;		int ret = 0;
	int bytes;		int bytes;
	@@ -373,14 +382,18 @@ int camogm_end_jpeg(camogm_state *state)
	if (state->rawdev_op) {		if (state->rawdev_op) {
	// write any remaining data, do not use writer thread as there can be only one block left CHUNK_REM buffer		// write any remaining data, do not use writer thread as there can be only one block left CHUNK_REM buffer
	pthread_mutex_lock(&state->writer_params.writer_mutex);		pthread_mutex_lock(&state->writer_params.writer_mutex);
	while (state->writer_params.data_ready)
			// while (state->writer_params.data_ready)
			while (state->writer_params.chunk_page_prep != state->writer_params.chunk_page_write)
	// wait for previous frame to be recorded first if it has not be recorded by the moment		// wait for previous frame to be recorded first if it has not be recorded by the moment
	pthread_cond_wait(&state->writer_params.main_cond, &state->writer_params.writer_mutex);		pthread_cond_wait(&state->writer_params.main_cond, &state->writer_params.writer_mutex);
	bytes = prep_last_block(state);
			bytes = prep_last_block(state, state->writer_params.chunk_page_prep % FILE_CHUNKS_PAGES);
	if (bytes > 0) {		if (bytes > 0) {
	D6(fprintf(debug_file, "Write last block of data, size = %d\n", bytes));		D6(fprintf(debug_file, "Write last block of data, size = %d\n", bytes));
	// the remaining data block is placed in CHUNK_COMMON buffer, write just this buffer		// the remaining data block is placed in CHUNK_COMMON buffer, write just this buffer
	iovlen = writev(state->writer_params.blockdev_fd, &state->writer_params.data_chunks[CHUNK_COMMON], 1);		iovlen = writev(state->writer_params.blockdev_fd,
			&state->writer_params.data_chunks[CHUNK_COMMON + (state->writer_params.chunk_page_prep % FILE_CHUNKS_PAGES) * FILE_CHUNKS_NUM], 1);
	if (iovlen < bytes) {		if (iovlen < bytes) {
	D0(fprintf(debug_file, "writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, bytes));		D0(fprintf(debug_file, "writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, bytes));
	state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;		state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;
	@@ -389,7 +402,9 @@ int camogm_end_jpeg(camogm_state *state)
	state->writer_params.lba_current += 1;		state->writer_params.lba_current += 1;
	state->rawdev.total_rec_len += bytes;		state->rawdev.total_rec_len += bytes;
	}		}
	reset_chunks(state->writer_params.data_chunks, 1);		// reset_chunks(state->writer_params.data_chunks, 1);
			//*************** See if another buffer(s) must be cleared too? or even use chunk_page_write
			reset_chunks(state, 1, state->writer_params.chunk_page_prep % FILE_CHUNKS_PAGES);
	}		}
	pthread_mutex_unlock(&state->writer_params.writer_mutex);		pthread_mutex_unlock(&state->writer_params.writer_mutex);

	@@ -411,28 +426,46 @@ int camogm_end_jpeg(camogm_state *state)
	*/		*/
	void jpeg_writer(void thread_args)		void jpeg_writer(void thread_args)
	{		{
			int dbg_us1, dbg_us2, dbg_us; // save usec for difference
	int reset_rem;		int reset_rem;
	int chunk_index;		int chunk_index, i0, i1;
	ssize_t iovlen, l;		ssize_t iovlen, l, ll, l0, l1, blocks_write, wpage;
	bool process = true;		bool process = true, has_data_to_write = false;
	struct iovec chunks_iovec[FILE_CHUNKS_NUM];		struct iovec chunks_iovec[FILE_CHUNKS_PAGES][FILE_CHUNKS_NUM];
			struct iovec split_chunk;

	camogm_state state = (camogm_state )thread_args;		camogm_state state = (camogm_state )thread_args;
	struct writer_params *params = &state->writer_params;		struct writer_params *params = &state->writer_params;
	unsigned char dummy_buff[PHY_BLOCK_SIZE];		unsigned char dummy_buff[PHY_BLOCK_SIZE];

	memset((void )chunks_iovec, 0, sizeof(struct iovec) FILE_CHUNKS_NUM);		memset((void )chunks_iovec, 0, sizeof(struct iovec) FILE_CHUNKS_PAGES * FILE_CHUNKS_NUM);
	pthread_mutex_lock(&params->writer_mutex);		// pthread_mutex_lock(&params->writer_mutex);
	params->state = STATE_RUNNING;		params->state = STATE_RUNNING;
			D6(fprintf(debug_file, "_w000_: Before jpeg_writer loop @ %06d\n", get_fpga_usec(state->fd_fparmsall[0], 0))); // before the debug file is set !
	while (process) {		while (process) {
			/*
	while (!params->data_ready && !params->exit_thread) {		while (!params->data_ready && !params->exit_thread) {
	pthread_cond_wait(&params->writer_cond, &params->writer_mutex);		pthread_cond_wait(&params->writer_cond, &params->writer_mutex);
			} // get rid of params->data_ready
			*/
			pthread_mutex_lock(&params->writer_mutex); // first time before the debug file is set !
			D6(fprintf(debug_file, "_w00_: (%d, %d) @ %06d\n", params->chunk_page_prep, params->chunk_page_write, get_fpga_usec(state->fd_fparmsall[0], 0)));
			while ((params->chunk_page_prep == params->chunk_page_write) && !params->exit_thread) {
			pthread_cond_wait(&params->writer_cond, &params->writer_mutex);
	}		}

	if (params->exit_thread) {		if (params->exit_thread) {
	process = false;		process = false;
	}		}
	if (params->data_ready) {		has_data_to_write = params->chunk_page_prep != params->chunk_page_write;
			pthread_mutex_unlock(&params->writer_mutex);
			D6(fprintf(debug_file, "_w00.5_: (%d, %d) @ %06d\n", params->chunk_page_prep, params->chunk_page_write,get_fpga_usec(state->fd_fparmsall[0], 0)));
			// if (params->data_ready) {
			if (has_data_to_write) {
			D6(fprintf(debug_file, "_w01_: got write data (%d > %d) @ %06d\n", params->chunk_page_prep, params->chunk_page_write,get_fpga_usec(state->fd_fparmsall[0], 0)));
	/* dummy read cycle from (approximately) the beginning of previous frame;		/* dummy read cycle from (approximately) the beginning of previous frame;
	* this is a debug feature used to find disk errors */		* this is a debug feature used to find disk errors */
			/*//temporary disable for simplicity 05/21/2023
	if (state->writer_params.dummy_read) {		if (state->writer_params.dummy_read) {
	ssize_t data_len;		ssize_t data_len;
	off64_t curr_offset = lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR);		off64_t curr_offset = lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR);
	@@ -445,26 +478,102 @@ void jpeg_writer(void thread_args)
	}		}
	lseek64(state->writer_params.blockdev_fd, curr_offset, SEEK_SET);		lseek64(state->writer_params.blockdev_fd, curr_offset, SEEK_SET);
	}		}
			*/
	/* end of dummy read cycle */		/* end of dummy read cycle */

	l = 0;		l = 0;
	reset_rem = 0;		reset_rem = 0;
	state->writer_params.last_ret_val = 0;		state->writer_params.last_ret_val = 0;
	chunk_index = get_data_buffers(state, chunks_iovec, FILE_CHUNKS_NUM);		// inputs from state->writer_params.data_chunks, returns number of result chunks
			wpage = state->writer_params.chunk_page_write % FILE_CHUNKS_PAGES;
			chunk_index = get_data_buffers(state, chunks_iovec[wpage], FILE_CHUNKS_NUM, wpage);
			// assuming write is block-aligned
	if (chunk_index > 0) {		if (chunk_index > 0) {
	for (int i = 0; i < chunk_index; i++)		for (int i = 0; i < chunk_index; i++)
	l += chunks_iovec[i].iov_len;		l += chunks_iovec[wpage][i].iov_len;
	iovlen = writev(state->writer_params.blockdev_fd, chunks_iovec, chunk_index);		// Moved rollover here
			blocks_write = l / PHY_BLOCK_SIZE;
			if (state->writer_params.lba_current + blocks_write <= state->writer_params.lba_end) { // almost always - single write
			state->writer_params.lba_current += blocks_write;
			// D6(fprintf(debug_file, "_w02_: starting writev @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
			D6(dbg_us1=get_fpga_usec(state->fd_fparmsall[0], 0); fprintf(debug_file, "_w02_: starting writev @ %06d\n",dbg_us1));
			iovlen = writev(state->writer_params.blockdev_fd, chunks_iovec[wpage], chunk_index);
			// D6(fprintf(debug_file, "_w03_: finished writev @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
			D6(dbg_us=get_fpga_usec(state->fd_fparmsall[0], 0); fprintf(debug_file, "_w03_: finished writev @ %06d (+ %06d, +%06d)\n",dbg_us, (dbg_us-dbg_us1), (dbg_us-dbg_us2)); dbg_us2=dbg_us);
	if (iovlen < l) {		if (iovlen < l) {
	D0(fprintf(debug_file, "writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, l));		D0(fprintf(debug_file, "writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, l));
	state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;		state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;
	reset_rem = 1;		reset_rem = 1;
			lseek(state->writer_params.blockdev_fd,
			(state->writer_params.lba_current - state->writer_params.lba_start) * PHY_BLOCK_SIZE,
			SEEK_SET); // update as if written?
	} else {		} else {
	// update statistic		// update statistic
	state->rawdev.last_jpeg_size = l;		state->rawdev.last_jpeg_size = l;
	state->rawdev.total_rec_len += state->rawdev.last_jpeg_size;		state->rawdev.total_rec_len += state->rawdev.last_jpeg_size;
	D6(fprintf(debug_file, "Current position in block device: %lld\n", lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR)));		D6(fprintf(debug_file, "Current position in block device: %lld\n", lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR)));
	}		}
			} else { // Split into two writes: to the end of partition and startion from the beginning
			l0 = (state->writer_params.lba_end - state->writer_params.lba_current) * PHY_BLOCK_SIZE; // remaining to be written before rollover
			l1 = l - l0; // remaining to be written after rollover
			i1 = 0; // start of the second write
			if (l0 > 0) { // should be always
			ll = 0;
			for (i0 = 0; i0 < chunk_index; i0++){
			if (ll + chunks_iovec[wpage][i0].iov_len > l0){
			break;
			}
			ll += chunks_iovec[wpage][i0].iov_len;
			}
			// save split chunk
			split_chunk.iov_base = chunks_iovec[wpage][i0].iov_base;
			split_chunk.iov_len = chunks_iovec[wpage][i0].iov_len;
			i1 = i0; // index of the first chunk in a second write
			if (ll < l0) { // is it a partial chunk?
			chunks_iovec[wpage][i0].iov_len = l0 - ll;
			i0++; // include split chunk in the first write
			}
			// write the first group (before rollover)
			state->writer_params.lba_current = state->writer_params.lba_start;
			iovlen = writev(state->writer_params.blockdev_fd, chunks_iovec[wpage], i0);
			lseek(state->writer_params.blockdev_fd, 0, SEEK_SET); // regardless of error - rewind
			if (iovlen < l0) {
			D0(fprintf(debug_file, "(1 of 2) writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, l0));
			state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;
			reset_rem = 1;
			} else {
			// do not update statistic after the first writed, do it after the second
			D6(fprintf(debug_file, "(1 of 2) Current position in block device: %lld\n", lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR)));
			// update split chunk to use its second part, shift chunks
			if (l1 > 0) { // length to be written after rollover
			if (chunks_iovec[wpage][i1].iov_len != split_chunk.iov_len) { // it was split
			D0(fprintf(debug_file, "chunk %d was split, full length = %d, writtren = %d\n", i1, split_chunk.iov_len, chunks_iovec[wpage][i1].iov_len));
			chunks_iovec[wpage][i1].iov_base += chunks_iovec[wpage][i1].iov_len;
			chunks_iovec[wpage][i1].iov_len = split_chunk.iov_len - chunks_iovec[wpage][i1].iov_len;
			}
			// shift chunks
			for (i0 = i1; i0 < chunk_index; i0++){
			chunks_iovec[wpage][i0 - i1] = chunks_iovec[wpage][i0];
			}
			iovlen = writev(state->writer_params.blockdev_fd, chunks_iovec[wpage], chunk_index - i1);
			state->writer_params.lba_current = state->writer_params.lba_start + l1 / PHY_BLOCK_SIZE;
			if (iovlen < l1) {
			D0(fprintf(debug_file, "(2 of 2) writev error: %s (returned %i, expected %i)\n", strerror(errno), iovlen, l1));
			state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;
			reset_rem = 1;
			} else {
			// update statistic (for both parts of the split write)
			state->rawdev.last_jpeg_size = l;
			state->rawdev.total_rec_len += state->rawdev.last_jpeg_size;
			D6(fprintf(debug_file, "(1 of 2) Current position in block device: %lld\n", lseek64(state->writer_params.blockdev_fd, 0, SEEK_CUR)));
			}
			}
			}
			}
			}
			D6(fprintf(debug_file, "_w04: wrote data @ %06d\n",get_fpga_usec(state->fd_fparmsall[0], 0)));
			D6(fprintf(debug_file, "_w05_: Block device positions: start = %llu, current = %llu, end = %llu\n",
			state->writer_params.lba_start, state->writer_params.lba_current, state->writer_params.lba_end));
	} else {		} else {
	D0(fprintf(debug_file, "data vector mapping error: %d)\n", chunk_index));		D0(fprintf(debug_file, "data vector mapping error: %d)\n", chunk_index));
	state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;		state->writer_params.last_ret_val = -CAMOGM_FRAME_FILE_ERR;
	@@ -472,14 +581,25 @@ void jpeg_writer(void thread_args)
	}		}

	// release main thread		// release main thread
	reset_chunks(state->writer_params.data_chunks, reset_rem);		// reset_chunks(state->writer_params.data_chunks, reset_rem);
	params->data_ready = false;		reset_chunks(state, reset_rem, state->writer_params.chunk_page_write % FILE_CHUNKS_PAGES);
	pthread_cond_signal(&params->main_cond);
			pthread_mutex_lock(&params->writer_mutex);
			params->chunk_page_write = (params->chunk_page_write + 1) % (2 * FILE_CHUNKS_PAGES);
			// params->data_ready = false;
			if (params->chunk_page_prep == params->chunk_page_write) {
			pthread_cond_signal(&params->main_cond); // ********** Do not signal if can write more !
			D6(fprintf(debug_file, "_w06_: wrote data (%d , %d), buffer empty, signal main thread @ %06d\n", params->chunk_page_prep, params->chunk_page_write,get_fpga_usec(state->fd_fparmsall[0], 0)));
			} else {
			D6(fprintf(debug_file, "_w06a_: wrote data (%d , %d), has more to write, do not signal! @ %06d\n", params->chunk_page_prep, params->chunk_page_write,get_fpga_usec(state->fd_fparmsall[0], 0)));

			}
			pthread_mutex_unlock(&params->writer_mutex); // do not release untill waiting for writev() or waiting for data to write
	}		}
	}		}
	params->state = STATE_STOPPED;		params->state = STATE_STOPPED;
	pthread_mutex_unlock(&state->writer_params.writer_mutex);		// pthread_mutex_unlock(&state->writer_params.writer_mutex);
	D5(fprintf(debug_file, "Exit from recording thread\n"));		D5(fprintf(debug_file, "_w07_: Exit from recording thread\n")); // never?

	return NULL;		return NULL;
	}		}

Admin message