extras/gomemif/examples/responder.go

   1 /*
   2  *------------------------------------------------------------------
   3  * Copyright (c) 2020 Cisco and/or its affiliates.
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *------------------------------------------------------------------
  16  */
  17
  18 package main
  19
  20 import (
  21         "bufio"
  22         "flag"
  23         "fmt"
  24         "os"
  25         "strings"
  26         "sync"
  27
  28         "github.com/pkg/profile"
  29         "memif"
  30 )
  31
  32 func Disconnected(i *memif.Interface) error {
  33         fmt.Println("Disconnected: ", i.GetName())
  34
  35         data, ok := i.GetPrivateData().(*interfaceData)
  36         if !ok {
  37                 return fmt.Errorf("Invalid private data")
  38         }
  39         close(data.quitChan) // stop polling
  40         close(data.errChan)
  41         data.wg.Wait() // wait until polling stops, then continue disconnect
  42
  43         return nil
  44 }
  45
  46 func Connected(i *memif.Interface) error {
  47         fmt.Println("Connected: ", i.GetName())
  48
  49         data, ok := i.GetPrivateData().(*interfaceData)
  50         if !ok {
  51                 return fmt.Errorf("Invalid private data")
  52         }
  53         data.errChan = make(chan error, 1)
  54         data.quitChan = make(chan struct{}, 1)
  55         data.wg.Add(1)
  56
  57         go func(errChan chan<- error, quitChan <-chan struct{}, wg *sync.WaitGroup) {
  58                 defer wg.Done()
  59                 // allocate packet buffer
  60                 pkt := make([]byte, 2048)
  61                 // get rx queue
  62                 rxq0, err := i.GetRxQueue(0)
  63                 if err != nil {
  64                         errChan <- err
  65                         return
  66                 }
  67                 // get tx queue
  68                 txq0, err := i.GetTxQueue(0)
  69                 if err != nil {
  70                         errChan <- err
  71                         return
  72                 }
  73                 for {
  74                         select {
  75                         case <-quitChan: // channel closed
  76                                 return
  77                         default:
  78                                 // read packet from shared memory
  79                                 pktLen, err := rxq0.ReadPacket(pkt)
  80                                 if pktLen > 0 {
  81                                         // write packet to shared memory
  82                                         txq0.WritePacket(pkt[:pktLen])
  83                                 } else if err != nil {
  84                                         errChan <- err
  85                                         return
  86                                 }
  87                         }
  88                 }
  89         }(data.errChan, data.quitChan, &data.wg)
  90
  91         return nil
  92 }
  93
  94 type interfaceData struct {
  95         errChan  chan error
  96         quitChan chan struct{}
  97         wg       sync.WaitGroup
  98 }
  99
 100 func interractiveHelp() {
 101         fmt.Println("help - print this help")
 102         fmt.Println("start - start connecting loop")
 103         fmt.Println("show - print interface details")
 104         fmt.Println("exit - exit the application")
 105 }
 106
 107 func main() {
 108         cpuprof := flag.String("cpuprof", "", "cpu profiling output file")
 109         memprof := flag.String("memprof", "", "mem profiling output file")
 110         role := flag.String("role", "slave", "interface role")
 111         name := flag.String("name", "gomemif", "interface name")
 112         socketName := flag.String("socket", "", "control socket filename")
 113
 114         flag.Parse()
 115
 116         if *cpuprof != "" {
 117                 defer profile.Start(profile.CPUProfile, profile.ProfilePath(*cpuprof)).Stop()
 118         }
 119         if *memprof != "" {
 120                 defer profile.Start(profile.MemProfile, profile.ProfilePath(*memprof)).Stop()
 121         }
 122
 123         memifErrChan := make(chan error)
 124         exitChan := make(chan struct{})
 125
 126         var isMaster bool
 127         switch *role {
 128         case "slave":
 129                 isMaster = false
 130         case "master":
 131                 isMaster = true
 132         default:
 133                 fmt.Println("Invalid role")
 134                 return
 135         }
 136
 137         fmt.Println("GoMemif: Responder")
 138         fmt.Println("-----------------------")
 139
 140         socket, err := memif.NewSocket("gomemif_example", *socketName)
 141         if err != nil {
 142                 fmt.Println("Failed to create socket: ", err)
 143                 return
 144         }
 145
 146         data := interfaceData{}
 147         args := &memif.Arguments{
 148                 IsMaster:         isMaster,
 149                 ConnectedFunc:    Connected,
 150                 DisconnectedFunc: Disconnected,
 151                 PrivateData:      &data,
 152                 Name:             *name,
 153         }
 154
 155         i, err := socket.NewInterface(args)
 156         if err != nil {
 157                 fmt.Println("Failed to create interface on socket %s: %s", socket.GetFilename(), err)
 158                 goto exit
 159         }
 160
 161         // slave attempts to connect to control socket
 162         // to handle control communication call socket.StartPolling()
 163         if !i.IsMaster() {
 164                 fmt.Println(args.Name, ": Connecting to control socket...")
 165                 for !i.IsConnecting() {
 166                         err = i.RequestConnection()
 167                         if err != nil {
 168                                 /* TODO: check for ECONNREFUSED errno
 169                                  * if error is ECONNREFUSED it may simply mean that master
 170                                  * interface is not up yet, use i.RequestConnection()
 171                                  */
 172                                 fmt.Println("Faild to connect: ", err)
 173                                 goto exit
 174                         }
 175                 }
 176         }
 177
 178         go func(exitChan chan<- struct{}) {
 179                 reader := bufio.NewReader(os.Stdin)
 180                 for {
 181                         fmt.Print("gomemif# ")
 182                         text, _ := reader.ReadString('\n')
 183                         // convert CRLF to LF
 184                         text = strings.Replace(text, "\n", "", -1)
 185                         switch text {
 186                         case "help":
 187                                 interractiveHelp()
 188                         case "start":
 189                                 // start polling for events on this socket
 190                                 socket.StartPolling(memifErrChan)
 191                         case "show":
 192                                 fmt.Println("remote: ", i.GetRemoteName())
 193                                 fmt.Println("peer: ", i.GetPeerName())
 194                         case "exit":
 195                                 err = socket.StopPolling()
 196                                 if err != nil {
 197                                         fmt.Println("Failed to stop polling: ", err)
 198                                 }
 199                                 close(exitChan)
 200                                 return
 201                         default:
 202                                 fmt.Println("Unknown input")
 203                         }
 204                 }
 205         }(exitChan)
 206
 207         for {
 208                 select {
 209                 case <-exitChan:
 210                         goto exit
 211                 case err, ok := <-memifErrChan:
 212                         if ok {
 213                                 fmt.Println(err)
 214                         }
 215                 case err, ok := <-data.errChan:
 216                         if ok {
 217                                 fmt.Println(err)
 218                         }
 219                 default:
 220                         continue
 221                 }
 222         }
 223
 224 exit:
 225         socket.Delete()
 226         close(memifErrChan)
 227 }