fellow_log: Prevent starting with too little memory

Part of the solution to #54

src/Makefile.am

@@ -295,7 +295,9 @@ fellow_log_dbg_SOURCES = fellow_log_dbg.c \
 	fellow_task.h \
 	fellow_testenv.h \
 	fellow_testenv.c \
-	foreign/vend.h
+	foreign/vend.h \
+	pow2_units.h \
+	pow2_units.c
 
 fellow_log_dbg_SOURCES += $(io_ioctl_sources)
 

src/fellow_log.c

@@ -47,6 +47,7 @@
 #include "foreign/qdef.h"
 #include "vqueue.h"
 #include "bitf_segmentation.h"
+#include "pow2_units.h"
 
 #include "foreign/vend.h"
 
@@ -550,8 +551,9 @@ struct fellow_disk_data_hdr {
 #define FELLOW_DISK_DATA_HDR_CHK_LEN \
 	(sizeof(struct fellow_disk_data_hdr) - FELLOW_DISK_DATA_HDR_CHK_START)
 	struct fellow_disk_log_info	log_info;
+	uint64_t			objsize;
 
-	uint8_t			_pad[3959];
+	uint8_t			_pad[3951];
 	uint8_t			fht;
 };
 
@@ -1931,12 +1933,67 @@ fellow_io_log_submit(const struct fellow_fd *ffd, void *ioctx,
 	return (1);
 }
 
+/*
+ * this is approximate, because we account overhead towards
+ * objects, also used space does not exactly match the numbers
+ * from the log buffer
+ */
+
+static uint64_t
+fellow_objsize(struct fellow_fd *ffd)
+{
+	const struct dle_stats *s;
+	size_t obj, del, spc, sz;
+	unsigned u;
+
+	if (ffd->rewriting)
+		return (0);
+
+	spc = buddy_space(ffd->dskbuddy, 1);
+	sz  = buddy_size(ffd->dskbuddy);
+
+	AN(sz);
+	assert(sz >= spc);
+
+	sz -= spc;
+
+	if (pthread_mutex_trylock(&ffd->logmtx))
+		return (0);
+
+	for (u = 0; u < LOGREGIONS; u++) {
+		spc = ffd->log_info.log_region[u].size;
+		if (sz >= spc)
+			sz -= spc;
+		else
+			sz = 0;
+	}
+	s = &ffd->logbuf->dle_stats;
+	obj = s->obj_add;
+	del = s->obj_del_free;
+	if (ffd->rewriting)
+		sz = 0;
+	AZ(pthread_mutex_unlock(&ffd->logmtx)); //lint !e455
+
+	spc = N_HDR * MIN_FELLOW_BLOCK;
+	if (sz >= spc)
+		sz -= spc;
+
+	if (obj > del)
+		obj -= del;
+
+	if (obj == 0)
+		return (0);
+
+	return (sz / obj);
+}
+
 static int
 fellow_io_write_hdr(struct fellow_fd *ffd)
 {
 	struct fellow_disk_data_hdr hdr
 	    __attribute__((aligned(MIN_FELLOW_BLOCK)));
 	int32_t ssz;
+	size_t objsz;
 
 	CHECK_OBJ_NOTNULL(ffd, FELLOW_FD_MAGIC);
 
@@ -1953,11 +2010,14 @@ fellow_io_write_hdr(struct fellow_fd *ffd)
 	    ffd->log_info.off.logblk,
 	    ffd->log_info.off.pendblk);
 
+	objsz = fellow_objsize(ffd);
+
 	INIT_OBJ(&hdr, FELLOW_DISK_DATA_HDR_MAGIC);
 	hdr.version = 1;
 	hdr.fht = FH_SHA256;
 	memcpy(&ffd->last_log_info, &ffd->log_info, sizeof hdr.log_info);
 	memcpy(&hdr.log_info, &ffd->log_info, sizeof hdr.log_info);
+	hdr.objsize = objsz;
 	fh(hdr.fht, hdr.fh,
 	    (char *)&hdr + FELLOW_DISK_DATA_HDR_CHK_START,
 	    FELLOW_DISK_DATA_HDR_CHK_LEN);
@@ -6299,7 +6359,8 @@ fellow_log_init(const char *path, size_t wantsize, size_t objsz_hint,
 	unsigned			i;
 	unsigned			entries;
 	size_t				sz;
-	int32_t			ssz;
+	int32_t				ssz;
+	uint64_t			objsize = 0;
 	char				last_err[256] = "";
 	const int *fl, tryflags[] = {
 #ifdef O_DIRECT
@@ -6437,6 +6498,8 @@ fellow_log_init(const char *path, size_t wantsize, size_t objsz_hint,
 
 		ffd->active_hdr = i;
 		memcpy(&ffd->log_info, &hdr[i].log_info, sizeof ffd->log_info);
+		if (hdr[i].objsize > 0)
+			objsize = hdr[i].objsize;
 	}
 
 	if (*last_err != '\0')
@@ -6456,6 +6519,22 @@ fellow_log_init(const char *path, size_t wantsize, size_t objsz_hint,
 		    path, ffd->active_hdr, ffd->log_info.generation);
 	}
 
+	if (objsize > MIN_FELLOW_BLOCK)
+		sz = MEMBUDDY_MINSIZE(wantsize, objsize);
+	else
+		sz = 0;
+
+	if (buddy_size(membuddy) < sz) {
+		ffd->diag("\nfellow:\tFATAL: memsz too small for "
+		    "actual objsize %zu < objsize_hint %zu!\n"
+		    "fellow:\tlower objsize_hint to %s or "
+		    "increase memsz to %s and restart.\n",
+		    (size_t)objsize, objsz_hint,
+		    pow2_unit(log2down(objsize)),
+		    pow2_unit(log2up(sz)));
+		goto err_close;
+	}
+
 	if (fstat(ffd->fd, &st)) {
 		ffd->diag("fellow: fstat(%s) failed: %s (%d)\n",
 		    path, strerror(errno), errno);
@@ -6561,6 +6640,7 @@ fellow_log_init(const char *path, size_t wantsize, size_t objsz_hint,
 
 	return (ffd);
   err_close:
+	DBG("ERROR close %p", ffd);
 	close(ffd->fd);
   err_free:
 	FREE_OBJ(ffd);

src/pow2_units.c

@@ -20,7 +20,11 @@
  * electronic and paper mail.
  */
 
-const char * const pow2_units[] = {
+#include "config.h"
+#include <assert.h>
+#include "pow2_units.h"
+
+const char * const pow2_units[POW2_MAX + 1] = {
 	[0] = "1B",
 	[1] = "2B",
 	[2] = "4B",

src/pow2_units.h

@@ -20,4 +20,13 @@
  * electronic and paper mail.
  */
 
-extern const char * const pow2_units[];
+#define POW2_MAX 80
+
+extern const char * const pow2_units[POW2_MAX + 1];
+
+static inline const char *
+pow2_unit(unsigned u)
+{
+	assert(u <= POW2_MAX);
+	return (pow2_units[u]);
+}