added chunks

rsload.go

@@ -9,6 +9,8 @@ import (
 	"runtime/pprof"
 )
 
+var chunk_size = 100
+
 var options struct {
 	host     string
 	port     int
@@ -77,8 +79,9 @@ func main() {
 		defer infile.Close()
 	}
 
+	type chunk []value
 	c := make(chan maybe)
-	sent := make(chan value, options.buffer)
+	sent := make(chan chunk, 200)
 	go streamValues(infile, c)
 	go func() {
 		w := bufio.NewWriterSize(conn, 16384)
@@ -86,27 +89,38 @@ func main() {
 			close(sent)
 			fmt.Println("All data transferred. Waiting for the last reply...")
 		}()
-		count := 0
+		requests := make(chunk, 0, chunk_size)
-		for r := range c {
+		for m := range c {
-			count++
+			if !m.ok() {
-			if r.ok() {
+				fmt.Fprintf(os.Stderr, "InputError: %v\n", m.err())
-				r.val().Write(w)
+				continue
-				if count == 100 {
+			}
-					w.Flush()
+			if _, err := m.val().Write(w); err != nil {
-					count = 0
+				fmt.Fprintf(os.Stderr, "WriteError: %v\n", err)
-				}
+			}
-				sent <- r.val()
+
-			} else {
+			requests = append(requests, m.val())
-				fmt.Fprintf(os.Stderr, "InputError: %v\n", r.err())
+			if len(requests) == cap(requests) {
-				return
+				if err := w.Flush(); err != nil {
+					fmt.Fprintf(os.Stderr, "FlushError: %v\n", err)
 				}
+				sent <- requests
+				requests = make(chunk, 0, chunk_size)
+			}
+		}
+		if len(requests) > 0 {
+			if err := w.Flush(); err != nil {
+				fmt.Fprintf(os.Stderr, "FlushError: %v\n", err)
+			}
+			sent <- requests
 		}
 	}()
 
 	replies, errors := 0, 0
 	responses := make(chan maybe)
 	go streamValues(conn, responses)
-	for request := range sent {
+	for requests := range sent {
+		for _, request := range requests {
 			response := <-responses
 			if response.ok() {
 				switch r := response.val().(type) {
@@ -127,6 +141,7 @@ func main() {
 				fmt.Fprintf(os.Stderr, "ResponseError: %v\n", response.err())
 			}
 		}
+	}
 	fmt.Println("Last reply received from server.")
 	fmt.Printf("errors: %d, replies: %d\n", errors, replies)
 }