<commit_before>package inmem

import (
	"fmt"
	"io"
	"sync"
)

type Handle interface {
	Send(msg *Message, mode int) (Handle, error)
	Receive(mode int) (*Message, Handle, error)
	CloseWrite() error
}

const (
	R	= 1 << (32 - 1 - iota)
	W
)

type Message struct {
	Target	string
	Name	string
	Args	[]string
	Data	string
}


func Pipe() (Handle, Handle) {
	red := make(chan *pipeMessage)
	black := make(chan *pipeMessage)
	return &PipeHandle{r: red, w: black}, &PipeHandle{r: black, w: red}
}

type PipeHandle struct {
	sync.RWMutex
	r chan *pipeMessage
	w chan *pipeMessage
}

func (h *PipeHandle) Send(msg *Message, mode int) (Handle, error) {
	h.RLock()
	defer h.RUnlock()
	if h.w == nil {
		return nil, fmt.Errorf("closed pipe")
	}
	var (
		rh Handle
		lh Handle
	)
	if mode&(R|W) != 0 {
		rh, lh = Pipe()
		if  mode&W == 0 {
			lh.CloseWrite()
		}
	}
	h.w <-&pipeMessage{msg, rh}
	return lh, nil
}

func (h *PipeHandle) Receive(mode int) (*Message, Handle, error) {
	pmsg, ok := <-h.r
	if !ok {
		return nil, nil, io.EOF
	}
	var handle Handle
	if pmsg.handle != nil && mode&W == 0 {
		pmsg.handle.CloseWrite()
	}
	if mode&(R|W) != 0 {
		handle = pmsg.handle
	}
	return pmsg.payload, handle, nil
}

func (h *PipeHandle) CloseWrite() error {
	h.Lock()
	defer h.Unlock()
	if h.w == nil {
		return fmt.Errorf("already closed")
	}
	close(h.w)
	h.w = nil
	return nil
}

type pipeMessage struct {
	payload *Message
	handle	Handle
}
<commit_msg>beam: no target field yet<commit_after>package inmem

import (
	"fmt"
	"io"
	"sync"
)

type Handle interface {
	Send(msg *Message, mode int) (Handle, error)
	Receive(mode int) (*Message, Handle, error)
	CloseWrite() error
}

const (
	R	= 1 << (32 - 1 - iota)
	W
)

type Message struct {
	Name	string
	Args	[]string
	Data	string
}


func Pipe() (Handle, Handle) {
	red := make(chan *pipeMessage)
	black := make(chan *pipeMessage)
	return &PipeHandle{r: red, w: black}, &PipeHandle{r: black, w: red}
}

type PipeHandle struct {
	sync.RWMutex
	r chan *pipeMessage
	w chan *pipeMessage
}

func (h *PipeHandle) Send(msg *Message, mode int) (Handle, error) {
	h.RLock()
	defer h.RUnlock()
	if h.w == nil {
		return nil, fmt.Errorf("closed pipe")
	}
	var (
		rh Handle
		lh Handle
	)
	if mode&(R|W) != 0 {
		rh, lh = Pipe()
		if  mode&W == 0 {
			lh.CloseWrite()
		}
	}
	h.w <-&pipeMessage{msg, rh}
	return lh, nil
}

func (h *PipeHandle) Receive(mode int) (*Message, Handle, error) {
	pmsg, ok := <-h.r
	if !ok {
		return nil, nil, io.EOF
	}
	var handle Handle
	if pmsg.handle != nil && mode&W == 0 {
		pmsg.handle.CloseWrite()
	}
	if mode&(R|W) != 0 {
		handle = pmsg.handle
	}
	return pmsg.payload, handle, nil
}

func (h *PipeHandle) CloseWrite() error {
	h.Lock()
	defer h.Unlock()
	if h.w == nil {
		return fmt.Errorf("already closed")
	}
	close(h.w)
	h.w = nil
	return nil
}

type pipeMessage struct {
	payload *Message
	handle	Handle
}

{'fraction_non_alphanumeric': 0.1518943343760862, 'fraction_numerical': 0.005561348627042058, 'mean_word_length': 3.360606060606061, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, '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': '5345183', 'n_tokens_mistral': 1234, 'n_tokens_neox': 1184, 'n_words': 342}