<commit_before>// Copyright 2015 Google Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

// #include "daemon.h"
// #include <linux/if_packet.h>
// #include <linux/filter.h>
// #include <stdlib.h>  // for C.free
import "C"

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"log"
	"net"
	"os/exec"
	"strconv"
	"sync"
	"time"
	"unsafe"
)

type Socket struct {
	num          int
	conf         SocketConfig
	fd           int
	newConns     chan *net.UnixConn
	oldConns     chan *conn
	newBlocks    chan *block
	blocks       []*block
	currentConns map[*conn]bool
	ring         uintptr
	index        int
}

func newSocket(sc SocketConfig, fanoutID int, num int) (*Socket, error) {
	s := &Socket{
		num:          num,
		conf:         sc,
		newConns:     make(chan *net.UnixConn),
		oldConns:     make(chan *conn),
		newBlocks:    make(chan *block),
		currentConns: map[*conn]bool{},
		blocks:       make([]*block, sc.NumBlocks),
	}
	var filt *C.struct_sock_fprog
	if sc.Filter != "" {
		f, err := compileFilter(sc.Interface, sc.Filter)
		if err != nil {
			return nil, fmt.Errorf("unable to compile filter %q on interface %q: %v", sc.Filter, sc.Interface, err)
		}
		filt = &f.filt
	}
	for i := 0; i < sc.NumBlocks; i++ {
		s.blocks[i] = &block{s: s, index: i}
	}
	iface := C.CString(sc.Interface)
	defer C.free(unsafe.Pointer(iface))
	var fd C.int
	var ring unsafe.Pointer
	var errStr *C.char
	if _, err := C.AFPacket(iface, C.int(sc.BlockSize), C.int(sc.NumBlocks),
		C.int(sc.BlockTimeoutMillis), C.int(fanoutID), C.int(sc.FanoutType), filt,
		&fd, &ring, &errStr); err != nil {
		return nil, fmt.Errorf("C AFPacket call failed: %v: %v", C.GoString(errStr), err)
	}
	s.fd = int(fd)
	s.ring = uintptr(ring)
	v(1, "%v set up with %+v", s, sc)
	return s, nil
}

func (s *Socket) String() string {
	return fmt.Sprintf("[S:%v:%v]", s.conf.SocketName, s.num)
}

func (s *Socket) getNewBlocks() {
	for {
		b := s.blocks[s.index]
		for !b.ready() {
			time.Sleep(time.Millisecond * 10)
		}
		b.ref()
		v(3, "%v got new block %v", s, b)
		s.newBlocks <- b
		s.index = (s.index + 1) % s.conf.NumBlocks
	}
}

func (s *Socket) run() {
	go s.getNewBlocks()
	for {
		select {
		case c := <-s.newConns:
			s.addNewConn(c)
		case c := <-s.oldConns:
			close(c.newBlocks)
			delete(s.currentConns, c)
		case b := <-s.newBlocks:
			for c, _ := range s.currentConns {
				b.ref()
				select {
				case c.newBlocks <- b:
				default:
					v(1, "failed to send %v to %v", b, c)
					b.unref()
				}
			}
			b.unref()
		}
	}
}

type conn struct {
	s         *Socket
	c         *net.UnixConn
	newBlocks chan *block
}

func (c *conn) String() string {
	return fmt.Sprintf("[C:%v:%v]", c.s, c.c.RemoteAddr())
}

// run handles communicating with a single external client via a single
// connection.  It maintains the invariant that every block it gets via the
// newBlocks channel will be unref'd exactly once.
func (c *conn) run() {
	outstanding := make([]time.Time, c.s.conf.NumBlocks)
	var mu sync.Mutex // protects outstanding
	readDone := make(chan struct{})
	writeDone := make(chan struct{})
	go func() { // handle reads
		defer close(readDone)
		for {
			var buf [4]byte
			n, err := c.c.Read(buf[:])
			if err == io.EOF {
				return
			} else if err != nil || n != len(buf) {
				v(1, "%v read error (%d bytes): %v", c, n, err)
				return
			}
			i := int(binary.BigEndian.Uint32(buf[:]))
			if i < 0 || i >= c.s.conf.NumBlocks {
				log.Printf("%v got invalid block %d", c, i)
				return
			}
			b := c.s.blocks[i]
			mu.Lock()
			t := outstanding[i]
			outstanding[i] = time.Time{}
			mu.Unlock()
			if t.IsZero() {
				log.Printf("%v returned %v that was not outstanding", c, b)
				return
			}
			v(4, "%v returned %v after %v", c, b, time.Since(t))
			b.unref()
			select {
			case <-writeDone:
				v(2, "%v read detected write closure", c)
				return
			default:
			}
		}<commit_msg>Log slow use of blocks.<commit_after>// Copyright 2015 Google Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

// #include "daemon.h"
// #include <linux/if_packet.h>
// #include <linux/filter.h>
// #include <stdlib.h>  // for C.free
import "C"

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"log"
	"net"
	"os/exec"
	"strconv"
	"sync"
	"time"
	"unsafe"
)

type Socket struct {
	num          int
	conf         SocketConfig
	fd           int
	newConns     chan *net.UnixConn
	oldConns     chan *conn
	newBlocks    chan *block
	blocks       []*block
	currentConns map[*conn]bool
	ring         uintptr
	index        int
}

func newSocket(sc SocketConfig, fanoutID int, num int) (*Socket, error) {
	s := &Socket{
		num:          num,
		conf:         sc,
		newConns:     make(chan *net.UnixConn),
		oldConns:     make(chan *conn),
		newBlocks:    make(chan *block),
		currentConns: map[*conn]bool{},
		blocks:       make([]*block, sc.NumBlocks),
	}
	var filt *C.struct_sock_fprog
	if sc.Filter != "" {
		f, err := compileFilter(sc.Interface, sc.Filter)
		if err != nil {
			return nil, fmt.Errorf("unable to compile filter %q on interface %q: %v", sc.Filter, sc.Interface, err)
		}
		filt = &f.filt
	}
	for i := 0; i < sc.NumBlocks; i++ {
		s.blocks[i] = &block{s: s, index: i}
	}
	iface := C.CString(sc.Interface)
	defer C.free(unsafe.Pointer(iface))
	var fd C.int
	var ring unsafe.Pointer
	var errStr *C.char
	if _, err := C.AFPacket(iface, C.int(sc.BlockSize), C.int(sc.NumBlocks),
		C.int(sc.BlockTimeoutMillis), C.int(fanoutID), C.int(sc.FanoutType), filt,
		&fd, &ring, &errStr); err != nil {
		return nil, fmt.Errorf("C AFPacket call failed: %v: %v", C.GoString(errStr), err)
	}
	s.fd = int(fd)
	s.ring = uintptr(ring)
	v(1, "%v set up with %+v", s, sc)
	return s, nil
}

func (s *Socket) String() string {
	return fmt.Sprintf("[S:%v:%v]", s.conf.SocketName, s.num)
}

func (s *Socket) getNewBlocks() {
	for {
		b := s.blocks[s.index]
		for !b.ready() {
			time.Sleep(time.Millisecond * 10)
		}
		b.ref()
		v(3, "%v got new block %v", s, b)
		s.newBlocks <- b
		s.index = (s.index + 1) % s.conf.NumBlocks
	}
}

func (s *Socket) run() {
	go s.getNewBlocks()
	for {
		select {
		case c := <-s.newConns:
			s.addNewConn(c)
		case c := <-s.oldConns:
			close(c.newBlocks)
			delete(s.currentConns, c)
		case b := <-s.newBlocks:
			for c, _ := range s.currentConns {
				b.ref()
				select {
				case c.newBlocks <- b:
				default:
					v(1, "failed to send %v to %v", b, c)
					b.unref()
				}
			}
			b.unref()
		}
	}
}

type conn struct {
	s         *Socket
	c         *net.UnixConn
	newBlocks chan *block
}

func (c *conn) String() string {
	return fmt.Sprintf("[C:%v:%v]", c.s, c.c.RemoteAddr())
}

// run handles communicating with a single external client via a single
// connection.  It maintains the invariant that every block it gets via the
// newBlocks channel will be unref'd exactly once.
func (c *conn) run() {
	outstanding := make([]time.Time, c.s.conf.NumBlocks)
	var mu sync.Mutex // protects outstanding
	readDone := make(chan struct{})
	writeDone := make(chan struct{})
	go func() { // handle reads
		defer close(readDone)
		for {
			var buf [4]byte
			n, err := c.c.Read(buf[:])
			if err == io.EOF {
				return
			} else if err != nil || n != len(buf) {
				v(1, "%v read error (%d bytes): %v", c, n, err)
				return
			}
			i := int(binary.BigEndian.Uint32(buf[:]))
			if i < 0 || i >= c.s.conf.NumBlocks {
				log.Printf("%v got invalid block %d", c, i)
				return
			}
			b := c.s.blocks[i]
			mu.Lock()
			t := outstanding[i]
			outstanding[i] = time.Time{}
			mu.Unlock()
			if t.IsZero() {
				log.Printf("%v returned %v that was not outstanding", c, b)
				return
			}
			duration := time.Since(t)
			level := 4
			if duration > time.Second / 4 {
				level = 1
			}
			v(level, "%v returned %v after %v", c, b, duration)
			b.unref()
			select {
			case <-writeDone:
				v(2, "%v read detected write closure", c)
				return
			default:
			}
		}
{'fraction_non_alphanumeric': 0.15569196428571427, 'fraction_numerical': 0.005133928571428571, 'mean_word_length': 2.7698779974758097, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2063341', 'n_tokens_mistral': 3671, 'n_tokens_neox': 3327, 'n_words': 1002}