read-ahead: handle memory allocation failures

Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

3 files changed, 111 insertions, 17 deletions

diff --git a/xlators/performance/read-ahead/src/page.c b/xlators/performance/read-ahead/src/page.c
index f9117dca506..7a311fc3f07 100644
--- a/xlators/performance/read-ahead/src/page.c
+++ b/xlators/performance/read-ahead/src/page.c
@@ -90,8 +90,10 @@ ra_wait_on_page (ra_page_t *page, call_frame_t *frame)
 	waitq = CALLOC (1, sizeof (*waitq));
 	if (!waitq) {
 		gf_log (frame->this->name, GF_LOG_ERROR,
-			"out of memory :(");
-		return;
+			"out of memory");
+                local->op_ret = -1;
+                local->op_errno = ENOMEM;
+                goto out;
 	}
 
 	waitq->data = frame;
@@ -103,6 +105,9 @@ ra_wait_on_page (ra_page_t *page, call_frame_t *frame)
 		local->wait_count++;
 	}
 	ra_local_unlock (local);
+
+out:
+        return;
 }
 
 
@@ -175,6 +180,11 @@ ra_fault_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		}
 
 		page->vector = iov_dup (vector, count);
+                if (page->vector == NULL) {
+                        waitq = ra_page_error (page, -1, ENOMEM);
+                        goto unlock;
+                }
+
 		page->count = count;
 		page->iobref = iobref_ref (iobref);
 		page->ready = 1;
@@ -202,10 +212,26 @@ void
 ra_page_fault (ra_file_t *file, call_frame_t *frame, off_t offset)
 {
 	call_frame_t *fault_frame = NULL;
-	ra_local_t   *fault_local = NULL;
+	ra_local_t   *fault_local = NULL, *local = NULL;
+        ra_page_t    *page = NULL;
+        ra_waitq_t   *waitq = NULL;
+        int32_t      op_ret = -1, op_errno = -1; 
     
+        local = frame->local;
 	fault_frame = copy_frame (frame);
+        if (fault_frame == NULL) {
+                op_ret = -1;
+                op_errno = ENOMEM;
+                goto err;
+        }
+
 	fault_local = CALLOC (1, sizeof (ra_local_t));
+        if (fault_local == NULL) {
+                STACK_DESTROY (fault_frame->root);
+                op_ret = -1;
+                op_errno = ENOMEM;
+                goto err;
+        }
 
 	fault_frame->local = fault_local;
 	fault_local->pending_offset = offset;
@@ -217,7 +243,22 @@ ra_page_fault (ra_file_t *file, call_frame_t *frame, off_t offset)
 		    FIRST_CHILD (fault_frame->this),
 		    FIRST_CHILD (fault_frame->this)->fops->readv,
 		    file->fd, file->page_size, offset);
+
 	return;
+
+err:
+        ra_file_lock (file);
+        {
+                page = ra_page_get (file, offset);
+                if (page)
+                        waitq = ra_page_error (page, op_ret,
+                                               op_errno);
+        }
+        ra_file_unlock (file);
+        
+        if (waitq != NULL) {
+                ra_waitq_return (waitq);
+        }
 }
 
 void
@@ -257,6 +298,11 @@ ra_frame_fill (ra_page_t *page, call_frame_t *frame)
 		}
 
 		new = CALLOC (1, sizeof (*new));
+                if (new == NULL) {
+                        local->op_ret = -1;
+                        local->op_errno = ENOMEM;
+                        goto out;
+                }
 
 		new->offset = page->offset;
 		new->size = copy_size;
@@ -265,6 +311,12 @@ ra_frame_fill (ra_page_t *page, call_frame_t *frame)
 					 src_offset, src_offset+copy_size,
 					 NULL);
 		new->vector = CALLOC (new->count, sizeof (struct iovec));
+                if (new->vector == NULL) {
+                        local->op_ret = -1;
+                        local->op_errno = ENOMEM;
+                        FREE (new);
+                        goto out;
+                }
 
 		new->count = iov_subset (page->vector, page->count,
 					 src_offset, src_offset+copy_size,
@@ -277,6 +329,9 @@ ra_frame_fill (ra_page_t *page, call_frame_t *frame)
 
 		local->op_ret += copy_size;
 	}
+
+out:
+        return;
 }
 
 
@@ -299,6 +354,10 @@ ra_frame_unwind (call_frame_t *frame)
 	fill  = local->fill.next;
 
 	iobref  = iobref_new ();
+        if (iobref == NULL) {
+                local->op_ret = -1;
+                local->op_errno = ENOMEM;
+        }
 
 	frame->local = NULL;
 
@@ -308,17 +367,25 @@ ra_frame_unwind (call_frame_t *frame)
 	}
 
 	vector = CALLOC (count, sizeof (*vector));
+        if (vector == NULL) {
+                local->op_ret = -1;
+                local->op_errno = ENOMEM;
+                iobref_unref (iobref);
+                iobref = NULL;
+        }
 
 	fill = local->fill.next;
 
 	while (fill != &local->fill) {
 		next = fill->next;
 
-		memcpy (((char *)vector) + copied, fill->vector,
-			fill->count * sizeof (*vector));
-
-		copied += (fill->count * sizeof (*vector));
-		iobref_merge (iobref, fill->iobref);
+                if ((vector != NULL) && (iobref != NULL)) {
+                        memcpy (((char *)vector) + copied, fill->vector,
+                                fill->count * sizeof (*vector));
+                        
+                        copied += (fill->count * sizeof (*vector));
+                        iobref_merge (iobref, fill->iobref);
+                }
 
 		fill->next->prev = fill->prev;
 		fill->prev->next = fill->prev;
diff --git a/xlators/performance/read-ahead/src/read-ahead.c b/xlators/performance/read-ahead/src/read-ahead.c
index b2703ad8d46..cb635e275b6 100644
--- a/xlators/performance/read-ahead/src/read-ahead.c
+++ b/xlators/performance/read-ahead/src/read-ahead.c
@@ -357,8 +357,11 @@ dispatch_requests (call_frame_t *frame, ra_file_t *file)
 				need_atime_update = 0;
 			}
 
-			if (!trav)
-				goto unlock;
+			if (!trav) {
+                                local->op_ret = -1;
+                                local->op_errno = ENOMEM;
+ 				goto unlock;
+                        }
 
 			if (trav->ready) {
 				gf_log (frame->this->name, GF_LOG_TRACE,
@@ -390,12 +393,17 @@ dispatch_requests (call_frame_t *frame, ra_file_t *file)
 		/* TODO: use untimens() since readv() can confuse underlying
 		   io-cache and others */
 		ra_frame = copy_frame (frame);
+                if (ra_frame == NULL) {
+                        goto out;
+                }
+
 		STACK_WIND (ra_frame, ra_need_atime_cbk,
 			    FIRST_CHILD (frame->this), 
 			    FIRST_CHILD (frame->this)->fops->readv,
 			    file->fd, 1, 1);
 	}
 
+out:
 	return ;
 }
 
@@ -493,8 +501,8 @@ ra_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
 
 	flush_region (frame, file, 0, floor (offset, file->page_size));
 
-	read_ahead (frame, file);
-
+        read_ahead (frame, file);
+                                 
 	ra_frame_return (frame);
 
 	file->offset = offset + size;
@@ -787,12 +795,13 @@ init (xlator_t *this)
 	ra_conf_t *conf;
 	dict_t    *options = this->options;
 	char      *page_count_string = NULL;
+        int32_t   ret = -1;
 
 	if (!this->children || this->children->next) {
 		gf_log (this->name,  GF_LOG_ERROR,
 			"FATAL: read-ahead not configured with exactly one"
                         " child");
-		return -1;
+                goto out;
 	}
 
 	if (!this->parents) {
@@ -801,7 +810,12 @@ init (xlator_t *this)
 	}
  
 	conf = (void *) CALLOC (1, sizeof (*conf));
-	ERR_ABORT (conf);
+        if (conf == NULL) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "FATAL: Out of memory");
+                goto out;
+        }
+
 	conf->page_size = this->ctx->page_size;
 	conf->page_count = 4;
 
@@ -818,7 +832,7 @@ init (xlator_t *this)
 				"invalid number format \"%s\" of \"option "
                                 "page-count\"", 
 				page_count_string);
-			return -1;
+                        goto out;
 		}
 		gf_log (this->name, GF_LOG_DEBUG, "Using conf->page_count = %u",
 			conf->page_count);
@@ -832,7 +846,7 @@ init (xlator_t *this)
 			gf_log (this->name, GF_LOG_ERROR,
 				"'force-atime-update' takes only boolean "
                                 "options");
-			return -1;
+                        goto out;
 		}
 		if (conf->force_atime_update)
 			gf_log (this->name, GF_LOG_DEBUG, "Forcing atime "
@@ -844,7 +858,16 @@ init (xlator_t *this)
 
 	pthread_mutex_init (&conf->conf_lock, NULL);
 	this->private = conf;
-	return 0;
+        ret = 0;
+
+out:
+        if (ret == -1) {
+                if (conf != NULL) {
+                        FREE (conf);
+                }
+        }
+
+        return ret;
 }
 
 void
diff --git a/xlators/performance/read-ahead/src/read-ahead.h b/xlators/performance/read-ahead/src/read-ahead.h
index 5513b269063..0f5add75ba5 100644
--- a/xlators/performance/read-ahead/src/read-ahead.h
+++ b/xlators/performance/read-ahead/src/read-ahead.h
@@ -124,9 +124,11 @@ typedef struct ra_fill ra_fill_t;
 ra_page_t *
 ra_page_get (ra_file_t *file,
 	     off_t offset);
+
 ra_page_t *
 ra_page_create (ra_file_t *file,
 		off_t offset);
+
 void
 ra_page_fault (ra_file_t *file,
 	       call_frame_t *frame,
@@ -134,6 +136,7 @@ ra_page_fault (ra_file_t *file,
 void
 ra_wait_on_page (ra_page_t *page,
 		 call_frame_t *frame);
+
 ra_waitq_t *
 ra_page_wakeup (ra_page_t *page);
 
@@ -149,6 +152,7 @@ ra_page_purge (ra_page_t *page);
 
 void
 ra_frame_return (call_frame_t *frame);
+
 void
 ra_frame_fill (ra_page_t *page,
 	       call_frame_t *frame);