diff --git a/emitter.go b/emitter.go
index 3db337c..1c985f1 100644
--- a/emitter.go
+++ b/emitter.go
@@ -152,7 +152,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				}
 			} else {
 				for _, dst := range writers {
-					if _, err := dst.PluginWrite(msg); err != nil {
+					if _, err := dst.PluginWrite(msg); err != nil && err != io.ErrClosedPipe {
 						return err
 					}
 				}
diff --git a/input_file.go b/input_file.go
index 5447492..d1e4230 100644
--- a/input_file.go
+++ b/input_file.go
@@ -4,9 +4,12 @@ import (
 	"bufio"
 	"bytes"
 	"compress/gzip"
+	"container/heap"
 	"errors"
+	"expvar"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -20,27 +23,68 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 )
 
-type fileInputReader struct {
-	reader    *bufio.Reader
+type filePayload struct {
 	data      []byte
-	file      io.ReadCloser
 	timestamp int64
-	closed    int32 // Value of 0 indicates that the file is still open.
-	s3        bool
 }
 
-func (f *fileInputReader) parseNext() error {
+// An IntHeap is a min-heap of ints.
+type payloadQueue struct {
+	sync.RWMutex
+	s []*filePayload
+}
+
+func (h payloadQueue) Len() int           { return len(h.s) }
+func (h payloadQueue) Less(i, j int) bool { return h.s[i].timestamp < h.s[j].timestamp }
+func (h payloadQueue) Swap(i, j int)      { h.s[i], h.s[j] = h.s[j], h.s[i] }
+
+func (h *payloadQueue) Push(x interface{}) {
+	// Push and Pop use pointer receivers because they modify the slice's length,
+	// not just its contents.
+	h.s = append(h.s, x.(*filePayload))
+}
+
+func (h *payloadQueue) Pop() interface{} {
+	old := h.s
+	n := len(old)
+	x := old[n-1]
+	h.s = old[0 : n-1]
+	return x
+}
+
+func (h payloadQueue) Idx(i int) *filePayload {
+	return h.s[i]
+}
+
+type fileInputReader struct {
+	reader    *bufio.Reader
+	file      io.ReadCloser
+	closed    int32 // Value of 0 indicates that the file is still open.
+	s3        bool
+	queue     payloadQueue
+	readDepth int
+}
+
+func (f *fileInputReader) parse(init chan struct{}) error {
 	payloadSeparatorAsBytes := []byte(payloadSeparator)
 	var buffer bytes.Buffer
+	var initialized bool
+
 	for {
 		line, err := f.reader.ReadBytes('\n')
 
 		if err != nil {
 			if err != io.EOF {
 				Debug(1, err)
-			} else {
-				f.Close()
 			}
+
+			f.Close()
+
+			if !initialized {
+				close(init)
+				initialized = true
+			}
+
 			return err
 		}
 
@@ -48,21 +92,51 @@ func (f *fileInputReader) parseNext() error {
 			asBytes := buffer.Bytes()
 			meta := payloadMeta(asBytes)
 
-			f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64)
-			f.data = asBytes[:len(asBytes)-1]
+			timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64)
+			data := asBytes[:len(asBytes)-1]
 
-			return nil
+			f.queue.Lock()
+			heap.Push(&f.queue, &filePayload{
+				timestamp: timestamp,
+				data:      data,
+			})
+			f.queue.Unlock()
+
+			for {
+				if f.queue.Len() < f.readDepth {
+					break
+				}
+
+				if !initialized {
+					close(init)
+					initialized = true
+				}
+
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			buffer = bytes.Buffer{}
+			continue
 		}
 
 		buffer.Write(line)
 	}
-
 }
 
-func (f *fileInputReader) ReadPayload() []byte {
-	defer f.parseNext()
+func (f *fileInputReader) wait() {
+	for {
+		if atomic.LoadInt32(&f.closed) == 1 {
+			return
+		}
 
-	return f.data
+		if f.queue.Len() > 0 {
+			return
+		}
+
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	return
 }
 
 // Close closes this plugin
@@ -75,7 +149,7 @@ func (f *fileInputReader) Close() error {
 	return nil
 }
 
-func newFileInputReader(path string) *fileInputReader {
+func newFileInputReader(path string, readDepth int) *fileInputReader {
 	var file io.ReadCloser
 	var err error
 
@@ -90,7 +164,7 @@ func newFileInputReader(path string) *fileInputReader {
 		return nil
 	}
 
-	r := &fileInputReader{file: file, closed: 0}
+	r := &fileInputReader{file: file, closed: 0, readDepth: readDepth}
 	if strings.HasSuffix(path, ".gz") {
 		gzReader, err := gzip.NewReader(file)
 		if err != nil {
@@ -102,7 +176,11 @@ func newFileInputReader(path string) *fileInputReader {
 		r.reader = bufio.NewReader(file)
 	}
 
-	r.parseNext()
+	heap.Init(&r.queue)
+
+	init := make(chan struct{})
+	go r.parse(init)
+	<-init
 
 	return r
 }
@@ -116,16 +194,25 @@ type FileInput struct {
 	readers     []*fileInputReader
 	speedFactor float64
 	loop        bool
+	readDepth   int
+	dryRun      bool
+	maxWait     time.Duration
+
+	stats *expvar.Map
 }
 
 // NewFileInput constructor for FileInput. Accepts file path as argument.
-func NewFileInput(path string, loop bool) (i *FileInput) {
+func NewFileInput(path string, loop bool, readDepth int, maxWait time.Duration, dryRun bool) (i *FileInput) {
 	i = new(FileInput)
 	i.data = make(chan []byte, 1000)
 	i.exit = make(chan bool)
 	i.path = path
 	i.speedFactor = 1
 	i.loop = loop
+	i.readDepth = readDepth
+	i.stats = expvar.NewMap("file-" + path)
+	i.dryRun = dryRun
+	i.maxWait = maxWait
 
 	if err := i.init(); err != nil {
 		return
@@ -165,7 +252,6 @@ func (i *FileInput) init() (err error) {
 	} else if matches, err = filepath.Glob(i.path); err != nil {
 		Debug(0, "[INPUT-FILE] Wrong file pattern", i.path, err)
 		return
-
 	}
 
 	if len(matches) == 0 {
@@ -176,9 +262,11 @@ func (i *FileInput) init() (err error) {
 	i.readers = make([]*fileInputReader, len(matches))
 
 	for idx, p := range matches {
-		i.readers[idx] = newFileInputReader(p)
+		i.readers[idx] = newFileInputReader(p, i.readDepth)
 	}
 
+	i.stats.Add("reader_count", int64(len(matches)))
+
 	return nil
 }
 
@@ -189,6 +277,7 @@ func (i *FileInput) PluginRead() (*Message, error) {
 	case <-i.exit:
 		return nil, ErrorStopped
 	case buf := <-i.data:
+		i.stats.Add("read_from", 1)
 		msg.Meta, msg.Data = payloadMetaWithBody(buf)
 		return &msg, nil
 	}
@@ -201,11 +290,17 @@ func (i *FileInput) String() string {
 // Find reader with smallest timestamp e.g next payload in row
 func (i *FileInput) nextReader() (next *fileInputReader) {
 	for _, r := range i.readers {
-		if r == nil || atomic.LoadInt32(&r.closed) != 0 {
+		if r == nil {
 			continue
 		}
 
-		if next == nil || r.timestamp < next.timestamp {
+		r.wait()
+
+		if r.queue.Len() == 0 {
+			continue
+		}
+
+		if next == nil || r.queue.Idx(0).timestamp < next.queue.Idx(0).timestamp {
 			next = r
 			continue
 		}
@@ -217,6 +312,11 @@ func (i *FileInput) nextReader() (next *fileInputReader) {
 func (i *FileInput) emit() {
 	var lastTime int64 = -1
 
+	var maxWait, firstWait, minWait int64
+	minWait = math.MaxInt64
+
+	i.stats.Add("negative_wait", 0)
+
 	for {
 		select {
 		case <-i.exit:
@@ -236,19 +336,48 @@ func (i *FileInput) emit() {
 			}
 		}
 
+		reader.queue.RLock()
+		payload := heap.Pop(&reader.queue).(*filePayload)
+		i.stats.Add("total_counter", 1)
+		i.stats.Add("total_bytes", int64(len(payload.data)))
+		reader.queue.RUnlock()
+
 		if lastTime != -1 {
-			diff := reader.timestamp - lastTime
+			diff := payload.timestamp - lastTime
+
+			if firstWait == 0 {
+				firstWait = diff
+			}
 
 			if i.speedFactor != 1 {
 				diff = int64(float64(diff) / i.speedFactor)
 			}
 
+			if i.maxWait > 0 && diff > int64(i.maxWait) {
+				diff = int64(i.maxWait)
+			}
+
 			if diff >= 0 {
-				lastTime = reader.timestamp
-				time.Sleep(time.Duration(diff))
+				lastTime = payload.timestamp
+
+				if !i.dryRun {
+					time.Sleep(time.Duration(diff))
+				}
+
+				i.stats.Add("total_wait", diff)
+
+				if diff > maxWait {
+					maxWait = diff
+				}
+
+				if diff < minWait {
+					minWait = diff
+				}
+			} else {
+				i.stats.Add("negative_wait", 1)
 			}
 		} else {
-			lastTime = reader.timestamp
+			lastTime = payload.timestamp
 		}
 
 		// Recheck if we have exited since last check.
@@ -256,12 +385,30 @@ func (i *FileInput) emit() {
 		case <-i.exit:
 			return
 		default:
-			i.data <- reader.ReadPayload()
+			if !i.dryRun {
+				i.data <- payload.data
+			}
 		}
 	}
 
+	i.stats.Set("first_wait", time.Duration(firstWait))
+	i.stats.Set("max_wait", time.Duration(maxWait))
+	i.stats.Set("min_wait", time.Duration(minWait))
+
 	Debug(0, fmt.Sprintf("[INPUT-FILE] FileInput: end of file '%s'\n", i.path))
 
+	if i.dryRun {
+		fmt.Printf("Records found: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n",
+			i.stats.Get("total_counter"),
+			i.stats.Get("reader_count"),
+			i.stats.Get("total_bytes"),
+			i.stats.Get("max_wait"),
+			i.stats.Get("min_wait"),
+			i.stats.Get("first_wait"),
+			time.Duration(i.stats.Get("total_wait").(*expvar.Int).Value()),
+			i.stats.Get("negative_wait"),
+		)
+	}
 }
 
 // Close closes this plugin
diff --git a/input_file_test.go b/input_file_test.go
index 91b9bcc..0538e62 100644
--- a/input_file_test.go
+++ b/input_file_test.go
@@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) {
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 
 	for i := '1'; i <= '4'; i++ {
 		msg, _ := input.PluginRead()
@@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) {
 	file.Write([]byte("1 3 250000000\nrequest3"))
 	file.Write([]byte(payloadSeparator))
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, 0, false)
 
 	start := time.Now().UnixNano()
 	for i := 0; i < 3; i++ {
@@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) {
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 
 	for i := '1'; i <= '4'; i++ {
 		msg, _ := input.PluginRead()
@@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) {
 	file.Write([]byte(payloadSeparator))
 	file.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, 0, false)
 
 	// Even if we have just 2 requests in file, it should indifinitly loop
 	for i := 0; i < 1000; i++ {
@@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) {
 	name2 := output2.file.Name()
 	output2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 	for i := 0; i < 2000; i++ {
 		input.PluginRead()
 	}
@@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile {
 func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) {
 	wg := new(sync.WaitGroup)
 
-	input := NewFileInput(captureFile.Name(), false)
+	input := NewFileInput(captureFile.Name(), false, 100, 0, false)
 	output := NewTestOutput(func(msg *Message) {
 		callback(msg)
 		wg.Done()
diff --git a/input_raw_test.go b/input_raw_test.go
index d844f6a..969e5a7 100644
--- a/input_raw_test.go
+++ b/input_raw_test.go
@@ -233,10 +233,11 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 	conf := RAWInputConfig{
-		Engine:        capture.EnginePcap,
-		Expire:        time.Second,
-		Protocol:      ProtocolHTTP,
-		TrackResponse: true,
+		Engine:          capture.EnginePcap,
+		Expire:          time.Second,
+		Protocol:        ProtocolHTTP,
+		TrackResponse:   true,
+		AllowIncomplete: true,
 	}
 	input := NewRAWInput(originAddr, conf)
 
diff --git a/output_dummy.go b/output_dummy.go
index e28653f..3d67a59 100644
--- a/output_dummy.go
+++ b/output_dummy.go
@@ -24,6 +24,7 @@ func (i *DummyOutput) PluginWrite(msg *Message) (int, error) {
 	n += nn
 	nn, err = os.Stdout.Write(payloadSeparatorAsBytes)
 	n += nn
+
 	return n, err
 }
 
diff --git a/output_file_test.go b/output_file_test.go
index d68c183..3fd00d6 100644
--- a/output_file_test.go
+++ b/output_file_test.go
@@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) {
 	emitter.Close()
 
 	var counter int64
-	input2 := NewFileInput("/tmp/test_requests.gor", false)
+	input2 := NewFileInput("/tmp/test_requests.gor", false, 100, 0, false)
 	output2 := NewTestOutput(func(*Message) {
 		atomic.AddInt64(&counter, 1)
 		wg.Done()
diff --git a/output_http.go b/output_http.go
index 43c8240..cb0c0de 100644
--- a/output_http.go
+++ b/output_http.go
@@ -215,6 +215,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) {
 	if !isRequestPayload(msg.Meta) {
 		return
 	}
+
 	uuid := payloadID(msg.Meta)
 	start := time.Now()
 	resp, err := client.Send(msg.Data)
diff --git a/plugins.go b/plugins.go
index b1cef48..e6b1a1e 100644
--- a/plugins.go
+++ b/plugins.go
@@ -83,7 +83,6 @@ func (plugins *InOutPlugins) registerPlugin(constructor interface{}, options ...
 		plugins.Outputs = append(plugins.Outputs, w)
 	}
 	plugins.All = append(plugins.All, plugin)
-
 }
 
 // NewPlugins specify and initialize all available plugins
@@ -119,7 +118,7 @@ func NewPlugins() *InOutPlugins {
 	}
 
 	for _, options := range Settings.InputFile {
-		plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop)
+		plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileMaxWait, Settings.InputFileDryRun)
 	}
 
 	for _, path := range Settings.OutputFile {
diff --git a/s3_test.go b/s3_test.go
index fc48be5..a6b12f9 100644
--- a/s3_test.go
+++ b/s3_test.go
@@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) {
 		<-output.closeCh
 	}
 
-	input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false)
+	input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, 0, false)
 
 	buf := make([]byte, 1000)
 	for i := 0; i <= 19999; i++ {
diff --git a/settings.go b/settings.go
index 954aa0a..b992731 100644
--- a/settings.go
+++ b/settings.go
@@ -45,10 +45,13 @@ type AppSettings struct {
 	OutputTCPConfig TCPOutputConfig
 	OutputTCPStats  bool `json:"output-tcp-stats"`
 
-	InputFile        MultiOption `json:"input-file"`
-	InputFileLoop    bool        `json:"input-file-loop"`
-	OutputFile       MultiOption `json:"output-file"`
-	OutputFileConfig FileOutputConfig
+	InputFile          MultiOption   `json:"input-file"`
+	InputFileLoop      bool          `json:"input-file-loop"`
+	InputFileReadDepth int           `json:"input-file-read-depth"`
+	InputFileDryRun    bool          `json:"input-file-dry-run"`
+	InputFileMaxWait   time.Duration `json:"input-file-max-wait"`
+	OutputFile         MultiOption   `json:"output-file"`
+	OutputFileConfig   FileOutputConfig
 
 	InputRAW MultiOption `json:"input_raw"`
 	RAWInputConfig
@@ -113,6 +116,9 @@ func init() {
 
 	flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com")
 	flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.")
+	flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance")
+	flag.BoolVar(&Settings.InputFileDryRun, "input-file-dry-run", false, "Simulate reading from the data source without replaying it. You will get information about expected replay time, number of found records etc.")
+	flag.DurationVar(&Settings.InputFileMaxWait, "input-file-max-wait", 0, "Set the maximum time between requests. Can help in situations when you have too long periods between request, and you want to skip them. Example: --input-raw-max-wait 1s")
 
 	flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor")
 	flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.")
diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go
index e89d5f8..66f1000 100644
--- a/tcp/tcp_message.go
+++ b/tcp/tcp_message.go
@@ -9,108 +9,9 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/buger/goreplay/simpletime"
 	"github.com/buger/goreplay/size"
 )
 
-var bufferPool = NewBufferPool(1000, 1)
-
-type buf struct {
-	b       []byte
-	created time.Time
-	gc      bool
-}
-
-type bufPool struct {
-	buffers chan *buf
-	ttl     int
-}
-
-func NewBufferPool(max int, ttl int) *bufPool {
-	pool := &bufPool{
-		buffers: make(chan *buf, max),
-		ttl:     ttl,
-	}
-
-	// Ensure that memory released over time
-	go func() {
-		var released int
-		// GC
-		for {
-			for i := 0; i < 100; i++ {
-				select {
-				case c := <-pool.buffers:
-					if simpletime.Now.Sub(c.created) < time.Duration(ttl)*time.Second {
-						select {
-						case pool.buffers <- c:
-						default:
-							stats.Add("active_buffer_count", -1)
-							c.b = nil
-							c.gc = true
-							released++
-						}
-					} else {
-						stats.Add("active_buffer_count", -1)
-						// Else GC
-						c.b = nil
-						c.gc = true
-						released++
-					}
-				default:
-					break
-				}
-			}
-
-			bufPoolCount.Set(int64(len(pool.buffers)))
-			releasedCount.Set(int64(released))
-
-			time.Sleep(1000 * time.Millisecond)
-		}
-	}()
-
-	return pool
-}
-
-// Borrow a Client from the pool.
-func (p *bufPool) Get() *buf {
-	var c *buf
-	select {
-	case c = <-p.buffers:
-	default:
-		stats.Add("total_alloc_buffer_count", 1)
-		stats.Add("active_buffer_count", 1)
-
-		c = new(buf)
-		c.b = make([]byte, 1024)
-		c.created = simpletime.Now
-
-		// Use this technique to find if pool leaks, and objects get GCd
-		//
-		// runtime.SetFinalizer(c, func(p *buf) {
-		// 	if !p.gc {
-		// 		panic("Pool leak")
-		// 	}
-		// })
-	}
-	return c
-}
-
-// Return returns a Client to the pool.
-func (p *bufPool) Put(c *buf) {
-	select {
-	case p.buffers <- c:
-	default:
-		stats.Add("active_buffers", -1)
-		c.gc = true
-		c.b = nil
-		// if pool overloaded, let it go
-	}
-}
-
-func (p *bufPool) Len() int {
-	return len(p.buffers)
-}
-
 // Stats every message carry its own stats object
 type Stats struct {
 	LostData  int
@@ -130,7 +31,6 @@ type Message struct {
 	packets  []*Packet
 	parser   *MessageParser
 	feedback interface{}
-	dataBuf  *buf
 	Stats
 }
 
@@ -164,8 +64,6 @@ func (m *Message) UUID() []byte {
 }
 
 func (m *Message) add(packet *Packet) bool {
-	// fmt.Println("SEQ:", packet.Seq, " - ", len(packet.Payload))
-
 	// Skip duplicates
 	for _, p := range m.packets {
 		if p.Seq == packet.Seq {
@@ -228,20 +126,10 @@ func (m *Message) PacketData() [][]byte {
 
 // Data returns data in this message
 func (m *Message) Data() []byte {
-	m.dataBuf = bufferPool.Get()
+	var tmp []byte
+	tmp, _ = copySlice(tmp, m.PacketData()...)
 
-	// var totalLen int
-	// for _, p := range m.packets {
-	// 	totalLen += len(p.Payload)
-	// }
-	// tmp := make([]byte, totalLen)
-	var n int
-	if m.dataBuf == nil {
-		panic("asdsd")
-	}
-	m.dataBuf.b, n = copySlice(m.dataBuf.b, m.PacketData()...)
-
-	return m.dataBuf.b[:n]
+	return tmp
 }
 
 // SetProtocolState set feedback/data that can be used later, e.g with End or Start hint
@@ -264,10 +152,6 @@ func (m *Message) Finalize() {
 	for _, p := range m.packets {
 		packetPool.Put(p)
 	}
-
-	if m.dataBuf != nil {
-		bufferPool.Put(m.dataBuf)
-	}
 }
 
 // Emitter message handler