Recordio cloud and local interface

go/master/c/CMakeLists.txt

@@ -6,7 +6,7 @@ set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${PARENT_DIR}/cmake")
 
 project(cxx_go C Go)
 
-include(golang)
+#include(golang)
 include(flags)
 
 set(MASTER_LIB_NAME "paddle_master")

go/master/c/client.go

@@ -13,6 +13,7 @@ typedef int paddle_master_client;
 import "C"
 
 import (
+	"io"
 	"sync"
 	"unsafe"
 
@@ -84,11 +85,29 @@ func paddle_set_dataset(client C.paddle_master_client, path **C.char, size C.int
 	return C.PADDLE_MASTER_OK
 }
 
+// return value:
+//     0:ok
+//    -1:EOF
+//    -2:error
 //export paddle_next_record
 func paddle_next_record(client C.paddle_master_client, record **C.uchar) C.int {
 	c := get(client)
-	r := c.NextRecord()
+	r, err := c.NextRecord()
+	if err == io.EOF {
+		// EOF
+		*record = (*C.uchar)(nullPtr)
+		return -1
+	}
+
+	if err != nil {
+		// Error
+		// TODO: return the type of error?
+		*record = (*C.uchar)(nullPtr)
+		return -2
+	}
+
 	if len(r) == 0 {
+		// Empty record
 		*record = (*C.uchar)(nullPtr)
 		return 0
 	}

go/master/client.go

@@ -1,6 +1,7 @@
 package master
 
 import (
+	"io"
 	"os"
 	"time"
 
@@ -17,7 +18,12 @@ type Addresser interface {
 // Client is the client of the master server.
 type Client struct {
 	conn *connection.Conn
-	ch   chan []byte
+	ch   chan record
+}
+
+type record struct {
+	r   []byte
+	err error
 }
 
 // NewClient creates a new Client.
@@ -27,7 +33,7 @@ type Client struct {
 func NewClient(addr Addresser, bufSize int) *Client {
 	c := &Client{}
 	c.conn = connection.New()
-	c.ch = make(chan []byte, bufSize)
+	c.ch = make(chan record, bufSize)
 	go c.monitorMaster(addr)
 	go c.getRecords()
 	return c
@@ -52,17 +58,20 @@ func (c *Client) getRecords() {
 
 			s := recordio.NewRangeScanner(f, &chunk.Index, -1, -1)
 			for s.Scan() {
-				c.ch <- s.Record()
+				c.ch <- record{s.Record(), nil}
 			}
 
 			if s.Err() != nil {
+				c.ch <- record{nil, s.Err()}
 				log.Errorln(err, chunk.Path)
 			}
 
 			err = f.Close()
 			if err != nil {
 				log.Errorln(err)
 			}
+
+			c.ch <- record{nil, io.EOF}
 		}
 
 		// We treat a task as finished whenever the last data
@@ -132,6 +141,7 @@ func (c *Client) taskFinished(taskID int) error {
 //
 // NextRecord will block until the next record is available. It is
 // thread-safe.
-func (c *Client) NextRecord() []byte {
-	return <-c.ch
+func (c *Client) NextRecord() ([]byte, error) {
+	r := <-c.ch
+	return r.r, r.err
 }

go/master/client_test.go

@@ -2,6 +2,7 @@ package master_test
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/rpc"
@@ -69,13 +70,22 @@ func TestNextRecord(t *testing.T) {
 
 	for pass := 0; pass < 50; pass++ {
 		received := make(map[byte]bool)
-		for i := 0; i < total; i++ {
-			r := c.NextRecord()
+		for i := 0; i <= total; i++ {
+			r, err := c.NextRecord()
+			if err == io.EOF {
+				break
+			}
+
+			if err != nil {
+				t.Fatal(pass, i, "Read error:", err)
+			}
+
 			if len(r) != 1 {
-				t.Fatal("Length should be 1.", r)
+				t.Fatal(pass, i, "Length should be 1.", r)
 			}
+
 			if received[r[0]] {
-				t.Fatal("Received duplicate.", received, r)
+				t.Fatal(pass, i, "Received duplicate.", received, r)
 			}
 			received[r[0]] = true
 		}

python/paddle/v2/master/client.py

@@ -26,14 +26,27 @@ def set_dataset(self, paths):
             holder[idx] = c_ptr
         lib.paddle_set_dataset(self.c, holder, len(paths))
 
+    # return format: (record, errno)
+    # errno =  0: ok
+    #       = -1: EOF
+    #       < -1: error
     def next_record(self):
         p = ctypes.c_char_p()
         ret = ctypes.pointer(p)
         size = lib.paddle_next_record(self.c, ret)
+        if size == -1:
+            # EOF
+            return None, -1
+
+        if size < -1:
+            # Error
+            return None, size
+
         if size == 0:
             # Empty record
-            return ""
+            return "", 0
+
         record = ret.contents.value[:size]
         # Memory created from C should be freed.
         lib.mem_free(ret.contents)
-        return record
+        return record, 0

python/paddle/v2/reader/creator.py

@@ -57,22 +57,51 @@ def reader():
     return reader
 
 
-def recordio(path):
+def recordio_local(paths):
     """
-    Creates a data reader that outputs record one one by one from given recordio file
-    :path: path of recordio file
-    :returns: data reader of recordio file
+    Creates a data reader that outputs record one one by one 
+        from given local recordio fils path.
+    :path: path of recordio files.
+    :returns: data reader of recordio files.
     """
 
     import recordio as rec
 
     def reader():
-        f = rec.reader(path)
+        for i, path in enumerate(paths):
+            f = rec.reader(path)
+            while True:
+                r = f.read()
+                if r is None:
+                    break
+                yield r
+            f.close()
+
+    return reader
+
+def recordio(paths, addr="", buf_size=100):
+    """
+    Creates a data reader that outputs record one one by one 
+        from given local or cloud recordio path.
+    :path: path of recordio files.
+    :returns: data reader of recordio files.
+    """
+    import os
+    import paddle.v2.master.client as cloud
+
+    if "KUBERNETES_SERVICE_HOST" not in os.environ.keys():
+        return recordio_local(paths)
+
+    def reader():
+        c = cloud(addr, buf_size)
+        c.set_dataset(paths)
+
         while True:
-            r = f.read()
+            r, err = client.next_record()
             if r is None:
                 break
             yield r
-        f.close()
+
+        c.close()
 
     return reader

python/paddle/v2/reader/tests/creator_test.py

@@ -38,7 +38,7 @@ class TestRecordIO(unittest.TestCase):
     def test_recordio(self):
         path = os.path.join(
             os.path.dirname(__file__), "test_recordio_creator.dat")
-        reader = paddle.v2.reader.creator.recordio(path)
+        reader = paddle.v2.reader.creator.recordio([path])
         for idx, r in enumerate(reader()):
             self.assertSequenceEqual(r, str(idx))