<commit_before>// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package modules

import (
	"bytes"
	"io"
	"sync"
)

// queueRW implements a ReadWriteCloser backed by an unbounded in-memory
// buffer.
type queueRW struct {
	mu     sync.Mutex
	cond   *sync.Cond
	buf    bytes.Buffer
	closed bool
}

func newRW() io.ReadWriteCloser {
	q := &queueRW{}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (q *queueRW) Close() error {
	q.mu.Lock()
	defer q.mu.Unlock()
	defer q.cond.Broadcast()
	q.closed = true
	return nil
}

func (q *queueRW) Read(p []byte) (n int, err error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	for q.buf.Len() == 0 && !q.closed {
		q.cond.Wait()
	}
	return q.buf.Read(p)
}

func (q *queueRW) Write(p []byte) (n int, err error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	defer q.cond.Broadcast()
	return q.buf.Write(p)
}<commit_msg>modules: improve queueRW error handling<commit_after>// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package modules

import (
	"bytes"
	"errors"
	"io"
	"sync"
)

// queueRW implements a ReadWriteCloser backed by an unbounded in-memory
// buffer.
type queueRW struct {
	mu     sync.Mutex
	cond   *sync.Cond
	buf    bytes.Buffer
	closed bool
}

func newRW() io.ReadWriteCloser {
	q := &queueRW{}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (q *queueRW) Close() error {
	q.mu.Lock()
	defer q.mu.Unlock()
	if !q.closed {
		defer q.cond.Broadcast()
		q.closed = true
	}
	return nil
}

func (q *queueRW) Read(p []byte) (n int, err error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	for {
		if q.buf.Len() > 0 {
			return q.buf.Read(p)
		}
		if q.closed {
			return 0, io.EOF
		}
		q.cond.Wait()
	}
}

var errWriteOnClosedPipe = errors.New("write on closed pipe")

func (q *queueRW) Write(p []byte) (n int, err error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.closed {
		return 0, errWriteOnClosedPipe
	}
	defer q.cond.Broadcast()
	return q.buf.Write(p)
}
{'fraction_non_alphanumeric': 0.14022662889518414, 'fraction_numerical': 0.0056657223796034, 'mean_word_length': 3.146771037181996, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6478413', 'n_tokens_mistral': 848, 'n_tokens_neox': 815, 'n_words': 249}