return md->tx_queues[index];
}
-// Copy packet data into the selected buffer.
+// Copy packet data into the selected buffer with splitting when necessary
static void
-govpp_copy_packet_data(memif_buffer_t *buffers, int index, void *data, uint16_t size)
+govpp_copy_packet_data(memif_buffer_t *buffers, uint16_t allocated, int bufIndex, void *packetData, uint16_t packetSize)
{
- buffers[index].len = (size > buffers[index].len ? buffers[index].len : size);
- memcpy(buffers[index].data, data, (size_t)buffers[index].len);
+ int dataOffset = 0;
+
+ do {
+ buffers[bufIndex].len = (packetSize > buffers[bufIndex].len ? buffers[bufIndex].len : packetSize);
+ void * curData = (packetData + dataOffset);
+ memcpy(buffers[bufIndex].data, curData, (size_t)buffers[bufIndex].len);
+ dataOffset += buffers[bufIndex].len;
+ bufIndex += 1;
+ packetSize -= buffers[bufIndex].len;
+ } while(packetSize > 0 && bufIndex < allocated && buffers[bufIndex].flags & MEMIF_BUFFER_FLAG_NEXT > 0);
}
// Get packet data from the selected buffer.
return buffers[index].data;
}
+// Checks if memif buffer is chained
+static int
+govpp_is_buffer_chained(memif_buffer_t *buffers, int index)
+{
+ return buffers[index].flags & MEMIF_BUFFER_FLAG_NEXT;
+}
+
+// Allocate memif buffers and return pointer to next free buffer
+static int
+govpp_memif_buffer_alloc(memif_conn_handle_t conn, uint16_t qid,
+ memif_buffer_t * bufs, uint16_t offset, memif_buffer_t ** nextFreeBuf,
+ uint16_t count, uint16_t * count_out, uint16_t size)
+{
+ memif_buffer_t * offsetBufs = (bufs + offset);
+ int err = memif_buffer_alloc(conn, qid, offsetBufs, count, count_out, size);
+ *count_out += offset;
+ *nextFreeBuf = offsetBufs;
+ return err;
+}
+
*/
import "C"
// Rx/Tx queues
ringSize int // number of items in each ring
+ bufferSize int // max buffer size
stopQPollFd int // event file descriptor used to stop pollRxQueue-s
wg sync.WaitGroup // wait group for all pollRxQueue-s
rxQueueBufs []CPacketBuffers // an array of C-libmemif packet buffers for each RX queue
type CPacketBuffers struct {
buffers *C.memif_buffer_t
count int
+ rxChainBuf []RawPacketData
}
// Context is a global Go-libmemif runtime context.
wg sync.WaitGroup /* wait-group for pollEvents() */
}
+type txPacketBuffer struct {
+ packets []RawPacketData
+ size int
+}
+
var (
// logger used by the adapter.
log *logger.Logger
log2RingSize = 10
}
+ bufferSize := config.BufferSize
+ if bufferSize <= 0 {
+ bufferSize = 2048
+ }
+
// Create memif-wrapper for Go-libmemif.
memif = &Memif{
- MemifMeta: config.MemifMeta,
- callbacks: &MemifCallbacks{},
- ifIndex: context.nextMemifIndex,
- ringSize: 1 << log2RingSize,
+ MemifMeta: config.MemifMeta,
+ callbacks: &MemifCallbacks{},
+ ifIndex: context.nextMemifIndex,
+ ringSize: 1 << log2RingSize,
+ bufferSize: int(bufferSize),
}
// Initialize memif callbacks.
// Multiple TxBurst-s can run concurrently provided that each targets a different
// TX queue.
func (memif *Memif) TxBurst(queueID uint8, packets []RawPacketData) (count uint16, err error) {
- var sentCount C.uint16_t
- var allocated C.uint16_t
- var bufSize int
-
if len(packets) == 0 {
return 0, nil
}
return 0, ErrQueueID
}
- // The largest packet in the set determines the packet buffer size.
+ var bufCount int
+ var buffers []*txPacketBuffer
+ cQueueID := C.uint16_t(queueID)
+
for _, packet := range packets {
- if len(packet) > int(bufSize) {
- bufSize = len(packet)
+ packetLen := len(packet)
+ log.Debugf("%v - preparing packet with len %v", cQueueID, packetLen)
+
+ if packetLen > memif.bufferSize {
+ // Create jumbo buffer
+ buffer := &txPacketBuffer{
+ size: packetLen,
+ packets: []RawPacketData{packet},
+ }
+
+ buffers = append(buffers, buffer)
+
+ // Increment bufCount by number of splits in this jumbo
+ bufCount += (buffer.size + memif.bufferSize - 1) / memif.bufferSize
+ } else {
+ buffersLen := len(buffers)
+
+ // This is very first buffer so there is no data to append to, prepare empty one
+ if buffersLen == 0 {
+ buffers = []*txPacketBuffer{{}}
+ buffersLen = 1
+ }
+
+ lastBuffer := buffers[buffersLen-1]
+
+ // Last buffer is jumbo buffer, create new buffer
+ if lastBuffer.size > memif.bufferSize {
+ lastBuffer = &txPacketBuffer{}
+ buffers = append(buffers, lastBuffer)
+ }
+
+ // Determine buffer size by max packet size in buffer
+ if packetLen > lastBuffer.size {
+ lastBuffer.size = packetLen
+ }
+
+ lastBuffer.packets = append(lastBuffer.packets, packet)
+ bufCount += 1
}
}
// Reallocate Tx buffers if needed to fit the input packets.
- pb := memif.txQueueBufs[queueID]
- bufCount := len(packets)
+ log.Debugf("%v - total buffer to allocate count %v", cQueueID, bufCount)
+ pb := &memif.txQueueBufs[queueID]
if pb.count < bufCount {
newBuffers := C.realloc(unsafe.Pointer(pb.buffers), C.size_t(bufCount*int(C.sizeof_memif_buffer_t)))
if newBuffers == nil {
}
// Allocate ring slots.
- cQueueID := C.uint16_t(queueID)
- errCode := C.memif_buffer_alloc(memif.cHandle, cQueueID, pb.buffers, C.uint16_t(bufCount),
- &allocated, C.uint32_t(bufSize))
- err = getMemifError(int(errCode))
- if err == ErrNoBufRing {
- // Not enough ring slots, <count> will be less than bufCount.
- err = nil
- }
- if err != nil {
- return 0, err
- }
+ var allocated C.uint16_t
+ var subCount C.uint16_t
+ for _, buffer := range buffers {
+ packetCount := C.uint16_t(len(buffer.packets))
+ isJumbo := buffer.size > memif.bufferSize
+
+ log.Debugf("%v - trying to send max buff size %v, packets len %v, buffer len %v, jumbo %v",
+ cQueueID, buffer.size, len(buffer.packets), packetCount, isJumbo)
+
+ var nextFreeBuff *C.memif_buffer_t
+ startOffset := allocated
+ errCode := C.govpp_memif_buffer_alloc(memif.cHandle, cQueueID, pb.buffers, startOffset, &nextFreeBuff,
+ packetCount, &allocated, C.uint16_t(buffer.size))
+
+ err = getMemifError(int(errCode))
+ endEarly := err == ErrNoBufRing
+ if endEarly {
+ // Not enough ring slots, <count> will be less than packetCount.
+ err = nil
+ }
+ if err != nil {
+ return 0, err
+ }
+
+ // Copy packet data into the buffers.
+ nowAllocated := allocated - startOffset
+ toFill := nowAllocated
+ if !isJumbo {
+ // If this is not jumbo frame, only 1 packet needs to be copied each iteration
+ toFill = 1
+ }
+
+ // Iterate over all packets and try to fill them into allocated buffers
+ // If packet is jumbo frame, continue filling to allocated buffers until no buffer is left
+ for i, packet := range buffer.packets {
+ if i >= int(nowAllocated) {
+ // There was less allocated buffers than actual packet count so exit early
+ break
+ }
+
+ packetData := unsafe.Pointer(&packet[0])
+ C.govpp_copy_packet_data(nextFreeBuff, toFill, C.int(i), packetData, C.uint16_t(len(packet)))
+ }
+
+ if isJumbo && nowAllocated > 0 {
+ // If we successfully allocated required amount of buffers for entire jumbo to be sent
+ // simply sub entire amount of jumbo frame packets and leave only 1 so sender will think
+ // it only sent 1 packet so it does not need to know anything about jumbo frames
+ subCount += nowAllocated - 1
+ }
- // Copy packet data into the buffers.
- for i := 0; i < int(allocated); i++ {
- packetData := unsafe.Pointer(&packets[i][0])
- C.govpp_copy_packet_data(pb.buffers, C.int(i), packetData, C.uint16_t(len(packets[i])))
+ // If we do not have enough buffers left to allocate, simply end here to avoid packet loss and try
+ // to handle it next burst
+ if endEarly {
+ break
+ }
}
- errCode = C.memif_tx_burst(memif.cHandle, cQueueID, pb.buffers, allocated, &sentCount)
+ var sentCount C.uint16_t
+ errCode := C.memif_tx_burst(memif.cHandle, cQueueID, pb.buffers, allocated, &sentCount)
err = getMemifError(int(errCode))
if err != nil {
return 0, err
}
- count = uint16(sentCount)
- return count, nil
+ // Prevent negative values
+ realSent := uint16(sentCount) - uint16(subCount)
+ if subCount > sentCount {
+ sentCount = 0
+ }
+
+ log.Debugf("%v - sent %v total allocated buffs %v", cQueueID, sentCount, allocated)
+ return realSent, nil
}
// RxBurst is used to receive multiple packets in one call from a selected queue.
}
// Reallocate Rx buffers if needed to fit the output packets.
- pb := memif.rxQueueBufs[queueID]
+ pb := &memif.rxQueueBufs[queueID]
bufCount := int(count)
if pb.count < bufCount {
newBuffers := C.realloc(unsafe.Pointer(pb.buffers), C.size_t(bufCount*int(C.sizeof_memif_buffer_t)))
return packets, err
}
+ chained := len(pb.rxChainBuf) > 0
+ if chained {
+ // We had stored data from previous burst because last buffer in previous burst was chained
+ // so we need to continue appending to this data
+ packets = pb.rxChainBuf
+ pb.rxChainBuf = nil
+ }
+
// Copy packet data into the instances of RawPacketData.
for i := 0; i < int(recvCount); i++ {
var packetSize C.int
packetData := C.govpp_get_packet_data(pb.buffers, C.int(i), &packetSize)
- packets = append(packets, C.GoBytes(packetData, packetSize))
+ packetBytes := C.GoBytes(packetData, packetSize)
+
+ if chained {
+ // We have chained buffers, so start merging packet data with last read packet
+ prevPacket := packets[len(packets)-1]
+ packets[len(packets)-1] = append(prevPacket, packetBytes...)
+ } else {
+ packets = append(packets, packetBytes)
+ }
+
+ // Mark last buffer as chained based on property on current buffer so next buffers
+ // will try to append data to this one in case we got jumbo frame
+ chained = C.govpp_is_buffer_chained(pb.buffers, C.int(i)) > 0
}
if recvCount > 0 {
packets = nil
}
+ if chained {
+ // We did not had enough space to process all chained buffers to the end so simply tell
+ // reader that it should not process any packets here and save them for next burst
+ // to finish reading the buffer chain
+ pb.rxChainBuf = packets
+ packets = nil
+ err = ErrNoBuf
+ }
+
return packets, err
}
--- /dev/null
+// jumbo-frames is simple example how to send larger and larger jumbo packets with libmemif adapter. This is simple copy
+// of raw-data but with sending larger packets, so for more information read its code and docs.
+package main
+
+import (
+ "fmt"
+ "os"
+ "os/signal"
+ "sync"
+ "time"
+
+ "git.fd.io/govpp.git/extras/libmemif"
+)
+
+const (
+ Socket = "/tmp/jumbo-frames-example"
+ Secret = "secret"
+ ConnectionID = 1
+ NumQueues uint8 = 3
+)
+
+var wg sync.WaitGroup
+var stopCh chan struct{}
+
+func OnConnect(memif *libmemif.Memif) (err error) {
+ details, err := memif.GetDetails()
+ if err != nil {
+ fmt.Printf("libmemif.GetDetails() error: %v\n", err)
+ }
+ fmt.Printf("memif %s has been connected: %+v\n", memif.IfName, details)
+
+ stopCh = make(chan struct{})
+ var i uint8
+ for i = 0; i < uint8(len(details.RxQueues)); i++ {
+ wg.Add(1)
+ go ReadAndPrintPackets(memif, i)
+ }
+ for i = 0; i < uint8(len(details.TxQueues)); i++ {
+ wg.Add(1)
+ go SendPackets(memif, i)
+ }
+ return nil
+}
+
+func OnDisconnect(memif *libmemif.Memif) (err error) {
+ fmt.Printf("memif %s has been disconnected\n", memif.IfName)
+ close(stopCh)
+ wg.Wait()
+ return nil
+}
+
+func ReadAndPrintPackets(memif *libmemif.Memif, queueID uint8) {
+ defer wg.Done()
+
+ interruptCh, err := memif.GetQueueInterruptChan(queueID)
+ if err != nil {
+ switch err {
+ case libmemif.ErrQueueID:
+ fmt.Printf("libmemif.Memif.GetQueueInterruptChan() complains about invalid queue id!?")
+ default:
+ fmt.Printf("libmemif.Memif.GetQueueInterruptChan() error: %v\n", err)
+ }
+ return
+ }
+
+ counter := 0
+ for {
+ select {
+ case <-interruptCh:
+ counter++
+ for {
+ packets, err := memif.RxBurst(queueID, 10)
+ if err != nil {
+ fmt.Printf("libmemif.Memif.RxBurst() error: %v\n", err)
+ } else {
+ if len(packets) == 0 {
+ break
+ }
+ for _, packet := range packets {
+ fmt.Printf("Received packet queue=%d: %v in burst %d\n", queueID, len(packet), counter)
+ }
+ }
+ }
+ case <-stopCh:
+ return
+ }
+ }
+}
+
+func SendPackets(memif *libmemif.Memif, queueID uint8) {
+ defer wg.Done()
+
+ counter := 0
+ for {
+ select {
+ case <-time.After(3 * time.Second):
+ counter++
+ packetMul := counter % 100 + 1 // Limit max iterations to 100 to not go out of bounds
+ packets := []libmemif.RawPacketData{
+ make([]byte, 128*packetMul),
+ make([]byte, 256*packetMul),
+ make([]byte, 512*packetMul),
+ }
+ sent := 0
+ for {
+ count, err := memif.TxBurst(queueID, packets[sent:])
+ if err != nil {
+ fmt.Printf("libmemif.Memif.TxBurst() error: %v\n", err)
+ break
+ } else {
+ fmt.Printf("libmemif.Memif.TxBurst() has sent %d packets in burst %v.\n", count, counter)
+ sent += int(count)
+ if sent == len(packets) {
+ break
+ }
+ }
+ }
+ case <-stopCh:
+ return
+ }
+ }
+}
+
+func main() {
+ var isMaster = true
+ var appSuffix string
+ if len(os.Args) > 1 && (os.Args[1] == "--slave" || os.Args[1] == "-slave") {
+ isMaster = false
+ appSuffix = "-slave"
+ }
+
+ appName := "jumbo-frames" + appSuffix
+ fmt.Println("Initializing libmemif as ", appName)
+ err := libmemif.Init(appName)
+ if err != nil {
+ fmt.Printf("libmemif.Init() error: %v\n", err)
+ return
+ }
+ defer libmemif.Cleanup()
+
+ memifCallbacks := &libmemif.MemifCallbacks{
+ OnConnect: OnConnect,
+ OnDisconnect: OnDisconnect,
+ }
+
+ memifConfig := &libmemif.MemifConfig{
+ MemifMeta: libmemif.MemifMeta{
+ IfName: "memif1",
+ ConnID: ConnectionID,
+ SocketFilename: Socket,
+ Secret: Secret,
+ IsMaster: isMaster,
+ Mode: libmemif.IfModeEthernet,
+ },
+ MemifShmSpecs: libmemif.MemifShmSpecs{
+ NumRxQueues: NumQueues,
+ NumTxQueues: NumQueues,
+ BufferSize: 2048,
+ Log2RingSize: 10,
+ },
+ }
+
+ fmt.Printf("Callbacks: %+v\n", memifCallbacks)
+ fmt.Printf("Config: %+v\n", memifConfig)
+
+ memif, err := libmemif.CreateInterface(memifConfig, memifCallbacks)
+ if err != nil {
+ fmt.Printf("libmemif.CreateInterface() error: %v\n", err)
+ return
+ }
+ defer memif.Close()
+
+ sigChan := make(chan os.Signal, 1)
+ signal.Notify(sigChan, os.Interrupt)
+ <-sigChan
+}
Code Review / govpp.git / commitdiff