3
\$\begingroup\$

I wrote a camera streaming app for pizero.

Since the pizero is too weak to do any video encoding I am using it to just capture usb webcam frames (mjpeg format) and forward them over udp to a powerful pc that does the video encoding.

When the light conditions are low there is a lot of noise introduced to the frames and they grow in size, overflowing the udp write size limit.

To tackle the issue I wrote a func that splits the frame in parts and forwards the pieces to the encoder.

What I am looking for in this review is the following:

Performance - App runs on pizero where resources are limited. I am using buffered channels and I am not sure if I implemented them correctly.

Best practices

Hidden pitfalls - If any, point out if I shot myself in the foot.

Code contains the whole camera implementation plus the server side.

Please review both.

My programming level is beginner. This is my first project.

Camera:

package main

import (
    "bytes"
    "fmt"
    "log"
    "math"
    "net"
    "os/exec"
    "sync"
    "time"

    "labix.org/v2/mgo/bson"
)

const (
    FFMPEG_ENCODER = "192.168.178.200:8000"
    CAM_ADDR       = ":5000"
    PACKET         = 60000
)

type msg struct {
    Fragment     bool
    FragmentID   int
    LastFragment bool
    Data         []byte
}

type reporting struct {
    Fps       int
    Size      int
    Fragments float64
    Encoding  time.Duration
    Writing   time.Duration
}

func main() {
    conn, err := udpDial()
    if err != nil {
        log.Fatal(err)
    }

    var stderr bytes.Buffer

    //bash script with commands for interfacing with the camera.
    //script outputs to stdout.
    cmd := exec.Command("./v4l2")
    pipe, _ := cmd.StdoutPipe()
    defer pipe.Close()
    cmd.Stderr = &stderr

    if err := cmd.Start(); err != nil {
        fmt.Println(fmt.Sprint(err) + ": " + stderr.String())
    }

    //pizero - has issues serializing.
    buffer := make(chan [][]byte, 500)
    var wg sync.WaitGroup
    var statistic reporting
    wg.Add(3)

    go func() {
        //used for statistics
        var fps int
        start := time.Now()

        for {
            fps++
            if time.Since(start) >= time.Second {
                statistic.Fps = fps
                fps = 0
                //reset timer
                start = time.Now()
            }

            frame := make([]byte, 80000)
            n, err := pipe.Read(frame)
            if err != nil {
                pipe.Close()
                log.Fatal(err)
            }

            frame = frame[0:n]

            statistic.Size = len(frame)

            go processFrame(frame, buffer, &statistic)
        }
    }()

    //
    go func() {
        for {
            //time how long it takes pi0 to serialize
            start := time.Now()
            fragments := <-buffer
            for _, fragment := range fragments {
                _, err := conn.Write(fragment)
                if err != nil {
                    log.Println(err)
                }
            }
            statistic.Writing = time.Since(start)
        }

    }()

    go func() {
        ticker := time.NewTicker(time.Second)
        for _ = range ticker.C {
            fmt.Printf("\r%vfps, fsize: %v, fragments: %v, enctime: %v, writetime: %v, bufsize: %v ", statistic.Fps, statistic.Size, statistic.Fragments, statistic.Encoding, statistic.Writing, len(buffer))
        }
    }()
    wg.Wait()
}

func processFrame(frame []byte, buffer chan<- [][]byte, statistic *reporting) {
    start := time.Now()
    f := msg{}

    //UDP single packet size 65507 bytes.
    //Func implements frame fragmentation.
    //Frames bigger than the UDP packet size limit will be split.

    //Fragment -  indicates if the message sent to the server is fragmented.
    //LastFragment - set to true when sending the last fragment to the server.
    //FragmentID - fragment identifier.

    //UDP does not guarantee arrival of the packet. FragmentID is needed by the server
    //in order to identify if the received message is part of a fragment and if it belongs
    //to the fragmented message received.

    //On mismatch the server will drop fragments it currently has in memory.

    length := len(frame)

    if length < PACKET {
        f.Fragment = false
        f.FragmentID = 1
        f.LastFragment = true
        f.Data = frame

        fragment, err := bson.Marshal(f)
        if err != nil {
            log.Fatal(err)
        }
        statistic.Encoding = time.Since(start)

        packet := [][]byte{fragment}

        buffer <- packet
        return
    }

    //Find out in how many pieces the frame must be split
    pieces := float64(length) / float64(PACKET)
    //Returns the floating point remainder(if any).
    //`pieces` is often not a whole number
    r := math.Remainder(float64(length), float64(PACKET))
    remainder := int(r)
    //Number of pieces that fit the defined packet constant
    fragments := int(math.Floor(pieces))
    statistic.Fragments = pieces

    /*
        fmt.Println("Pieces: ", pieces)
        fmt.Println("Remainder: ", remainder)
        fmt.Println("Fragments: ", fragments)
        fmt.Println("Bytes: ", length)
    */

    //remainder can be a negative number
    //find out the positive remainder
    //remainder must be always positive
    //used for byte reslicing
    if remainder < 0 {
        remainder = PACKET + remainder //remainder here is a negative num
    }

    //default values
    id := 1
    beginning := 0
    end := remainder
    last := false

    //fragment storage. a go routine will loop over it and write to server
    packet := make([][]byte, int(math.Ceil(pieces)))

    for i := 0; fragments >= 0; i++ {
        f.Fragment = true
        f.FragmentID = id
        f.LastFragment = last
        f.Data = frame[beginning:end]

        fragment, err := bson.Marshal(f)
        if err != nil {
            log.Fatal(err)
        }

        packet[i] = fragment
        /*
            fmt.Printf("Start:%v, End:%v, Data length:%v,Packet length %v\n", start, end, len(f.Data), len(packet))
            fmt.Printf("Data length:%v,Packet length %v\n", len(f.Data), len(packet))
        */

        fragments--
        beginning = end
        end = beginning + PACKET

        if fragments == 0 {
            last = true
        }

        id++
    }
    buffer <- packet
    statistic.Encoding = time.Since(start)
}

func udpDial() (*net.UDPConn, error) {
    encAddr, err := net.ResolveUDPAddr("udp", FFMPEG_ENCODER)
    if err != nil {
        return nil, err
    }

    locAddr, err := net.ResolveUDPAddr("udp", CAM_ADDR)
    if err != nil {
        return nil, err
    }

    conn, err := net.DialUDP("udp", locAddr, encAddr)
    if err != nil {
        return nil, err
    }

    return conn, nil
}

Server:

package main

import (
    "bytes"
    "fmt"
    "log"
    "net"
    "os"
    "sync"

    "labix.org/v2/mgo/bson"
)

const (
    CONN_HOST = "192.168.178.200"
    CONN_PORT = "8000"
    CONN_TYPE = "udp"
)

type msg struct {
    Fragment     bool
    FragmentID   int
    LastFragment bool
    Data         []byte
}

func main() {
    logfile := "log.txt"
    logger, err := os.Create(logfile)
    if err != nil {
        fmt.Println(err)
    }

    log.SetOutput(logger)
    defer logger.Close()

    ladr, err := net.ResolveUDPAddr(CONN_TYPE, CONN_HOST+":"+CONN_PORT)
    if err != nil {
        fmt.Println(err)
        return
    }

    listen, err := net.ListenUDP(CONN_TYPE, ladr)
    if err != nil {
        fmt.Println(err)
        return
    }

    defer listen.Close()

    fmt.Println("Listening on " + CONN_HOST + ":" + CONN_PORT)

    collector := [][]byte{}

    var fragmentID int
    var expectedID int
    expectedID = 1

    buffer := make(chan msg, 1000)

    wg := sync.WaitGroup{}
    wg.Add(2)

    go func() {
        for {
            buf := make([]byte, 65507)
            n, _, err := listen.ReadFromUDP(buf)
            if err != nil {
                fmt.Println(err)
                break
            }

            if len(buf[0:n]) > 0 {
                f := msg{}

                err = bson.Unmarshal(buf[0:n], &f)
                if err != nil {
                    log.Fatal(err)
                }
                buffer <- f
            }
        }
    }()

    go func() {
        for {
            f := <-buffer
            //log.Printf("fragment :%v, fragId: %v, expectedID: %v, lastFragment: %v\n", f.Fragment, f.FragmentID, expectedID, f.LastFragment)
            //fmt.Printf("Data: %v\n", len(f.Data))
            if !f.Fragment {
                _, err = os.Stdout.Write(f.Data)
                if err != nil {
                    fmt.Println(err)
                }
                continue
            }

            if f.Fragment && !f.LastFragment {

                fragmentID = f.FragmentID

                if fragmentID == expectedID {
                    collector = append(collector, f.Data)
                    expectedID++
                    continue
                }
                log.Println("INCOMPLETE FRAME")
            }

            if f.Fragment && f.LastFragment {

                fragmentID = f.FragmentID

                if fragmentID == expectedID {
                    collector = append(collector, f.Data)

                    frame := bytes.Join(collector, []byte(""))

                    _, err = os.Stdout.Write(frame)
                    if err != nil {
                        fmt.Println(err)
                    }

                    //fmt.Println("Bytes: ", len(frame))

                    expectedID = 1
                    collector = collector[:0]

                    continue
                }
                log.Println("INCOMPLETE FRAME")

            }
            expectedID = 1
            collector = collector[:0]
        }
    }()
    wg.Wait()
}
\$\endgroup\$
2
\$\begingroup\$

Readability

To ease the readability (and testability) of your code, I would recommend you to split it in multiple functions communicating via multiple channels.

For instance, for the client:

func getOutput(command string) (io.ReadCloser, error)        // to launch the v4l2
func readFrames(in io.ReadCloser, out <-chan Frame)
func splitFrames(in chan<- Frame, out <-chan Fragment)
func sendFragments(in <-chan Fragment)

Your main would then consist of creating the different channels and launching those functions as goroutines (except for getOutput).

With this approach, you will need a fixed amount of goroutines (instead of starting a new one for each frame: go processFrame(frame, buffer, &statistic), which might qualify as "I shot myself in the foot").

Statistics

Regarding the statistics, if see multiple possibilities:

  • pass the object as a variable (current usage)
  • or make it a global variable
  • or make it a channel of Report (the signification of Value depends on the Operation):

-

type Report struct {
    Operation string
    Value     int
    Duration  time.Duration
}

Performance

If you need to make your code faster, you will need to identify the bottlenecks. For this, I recommend you to profile your code : https://blog.golang.org/profiling-go-programs

  • Add the code var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")...
  • transfer the *.profile file to your computer
  • analyze the function calls
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.