func (b *block) String() string {
	return fmt.Sprintf("[B:%v:%v]", b.s, b.index)
}
String provides a unique human-readable string.
package socket
import "C"
import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"log"
	"net"
	"os/exec"
	"strconv"
	"sync/atomic"
	"syscall"
	"time"
	"unsafe"
	"github.com/google/testimony/go/protocol"
	"github.com/google/testimony/go/testimonyd/internal/vlog"
)
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             
}
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, sc.NumBlocks),
		currentConns: map[*conn]bool{},
		blocks:       make([]*block, sc.NumBlocks),
	}
	var filt *C.struct_sock_filter
	var filtsize C.int
	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[0]
		filtsize = C.int(len(f))
	}
	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.FanoutSize), C.int(sc.FanoutType),
		filtsize, 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)
	log.Printf("%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() {
	blockIndex := 0
	for {
		b := s.blocks[blockIndex]
		for !b.ready() {
			if err := C.WaitForBlocks(C.int(s.fd)); err != 0 {
				log.Panicf("C WaitForBlocks failed: %s", syscall.Errno(err).Error())
			}
		}
		b.ref()
		vlog.V(3, "%v got new block %v", s, b)
		s.newBlocks <- b
		blockIndex = (blockIndex + 1) % s.conf.NumBlocks
	}
}
func (s *socket) reportStats() {
	var totalPackets, totalDrops uint64
	s.stats()
	const seconds = 60
	for range time.Tick(time.Second * seconds) {
		stats, err := s.stats()
		if err != nil {
			log.Printf("error getting statistics: %v", err)
		} else {
			totalPackets += uint64(stats.tp_packets)
			totalDrops += uint64(stats.tp_drops)
			vlog.V(1, "%v stats: %d packets (%.02fpps), %d drops (%.02fpps) (%.02f%% dropped) since last log, %d packets, %d drops total (%.02f%% dropped)", s,
				stats.tp_packets, float64(stats.tp_packets)/seconds, stats.tp_drops, float64(stats.tp_drops)/seconds, float64(stats.tp_drops)/float64(stats.tp_drops+stats.tp_packets)*100,
				totalPackets, totalDrops, float64(totalDrops)/float64(totalPackets+totalDrops)*100)
		}
	}
}
func (s *socket) run() {
	go func() {
		s.getNewBlocks()
	}()
	go s.reportStats()
	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:
					vlog.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
	oldBlocks chan int
}
func (c *conn) String() string {
	return fmt.Sprintf("[C:%v:%v]", c.s, c.c.RemoteAddr())
}
func (c *conn) handleReads() {
	defer close(c.oldBlocks)
	for {
		var buf [4]byte
		n, err := c.c.Read(buf[:])
		if err == io.EOF {
			return
		} else if err != nil || n != len(buf) {
			vlog.V(1, "%v read error (%d bytes): %v", c, n, err)
			return
		}
		num := binary.BigEndian.Uint32(buf[:])
		if num&0x80000000 != 0 {
			typ, length := protocol.TLFrom(num)
			if protocol.TypeOf(typ) != protocol.TypeClientToServer {
				vlog.V(1, "%v client sent bad type %d", c, typ)
			}
			var val []byte
			if length != 0 {
				val = make([]byte, length)
				if _, err := io.ReadFull(c.c, val); err != nil {
					vlog.V(2, "%v read TLV %d length %d: %v", c, typ, length, err)
					return
				}
			}
			if err := c.handleTLV(typ, val); err != nil {
				vlog.V(2, "%v handling type %d: %v", c, typ, val)
				return
			}
		} else {
			i := int(num)
			if i < 0 || i >= c.s.conf.NumBlocks {
				log.Printf("%v got invalid block %d", c, i)
				return
			}
			c.oldBlocks <- i + 1
		}
	}
}
func (c *conn) handleTLV(typ protocol.Type, val []byte) error {
	log.Printf("IGNORING TLV: %d = %x", typ, val)
	return nil
}
func (c *conn) run() {
	go c.handleReads()
	outstanding := make([]time.Time, len(c.s.blocks))
	var out []byte
loop:
	for {
		select {
		case b := <-c.newBlocks:
			out = out[:0]
			vlog.V(2, "%v writing %v", c, b)
		blockLoop:
			for {
				if !outstanding[b.index].IsZero() {
					log.Fatalf("%v received already outstanding block %v", c, b)
				}
				outstanding[b.index] = time.Now()
				idx := len(out)
				out = append(out, 0, 0, 0, 0)
				binary.BigEndian.PutUint32(out[idx:], uint32(b.index))
				select {
				case b = <-c.newBlocks:
					vlog.V(2, "%v batching %v", c, b)
				default:
					break blockLoop
				}
			}
			if _, err := c.c.Write(out); err != nil {
				vlog.V(1, "%v write error: %v", c, err)
				break loop
			}
		case i := <-c.oldBlocks:
			if i == 0 {
				break loop
			}
			i--  
			if outstanding[i].IsZero() {
				log.Printf("%v received non-outstanding block %v from client", c, i)
				break loop
			}
			b := c.s.blocks[i]
			vlog.V(3, "%v took %v to process block %v", c, time.Since(outstanding[i]), b)
			outstanding[i] = time.Time{}
			b.unref()  
		}
	}
	log.Printf("Connection %v closing", c)
	c.c.Close()
	vlog.V(3, "%v marking self old", c)
	c.s.oldConns <- c
	vlog.V(3, "%v waiting for reads", c)
	for b := range c.newBlocks {
		vlog.V(3, "%v returning unsent %v", c, b)
		b.unref()
	}
	for _ = range c.oldBlocks {
	}
	for i, t := range outstanding {
		if !t.IsZero() {
			b := c.s.blocks[i]
			vlog.V(3, "%v returning outstanding %v after %v", c, b, time.Since(t))
			b.unref()
		}
	}
}
func (s *socket) addNewConn(c *net.UnixConn) {
	newConn := &conn{
		s:         s,
		c:         c,
		newBlocks: make(chan *block, len(s.blocks)),
		oldBlocks: make(chan int, len(s.blocks)),
	}
	log.Printf("%v new connection %v", s, newConn)
	s.currentConns[newConn] = true
	go newConn.run()
}
type block struct {
	s     *socket
	index int  
	r int32  
}
func (b *block) ref() {
	refs := atomic.AddInt32(&b.r, 1)
	vlog.VUp(5, 1, "%v refs = %d", b, refs)
}
func (b *block) unref() {
	refs := atomic.AddInt32(&b.r, -1)
	vlog.VUp(5, 1, "%v unref = %d", b, refs)
	if refs == 0 {
		b.clear()
	} else if refs < 0 {
		panic(fmt.Sprintf("invalid unref of %v to %d", b, refs))
	}
}