<commit_before>package semtechudp

import (
	"crypto/rand"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"sync"
	"time"

	"github.com/pkg/errors"
	log "github.com/sirupsen/logrus"

	"github.com/brocaar/chirpstack-api/go/v3/gw"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/events"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/semtechudp/packets"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/filters"
	"github.com/brocaar/lorawan"
)

// udpPacket represents a raw UDP packet.
type udpPacket struct {
	addr *net.UDPAddr
	data []byte
}

type pfConfiguration struct {
	gatewayID      lorawan.EUI64
	baseFile       string
	outputFile     string
	restartCommand string
	currentVersion string
}

// Backend implements a Semtech packet-forwarder (UDP) gateway backend.
type Backend struct {
	sync.RWMutex

	// tokenMap stores the token to UUID mapping. This should take ~ 1MB of
	// memory. Optionaly this could be optimized by letting keys expire after
	// a given time.
	tokenMap map[uint16][]byte

	downlinkTXAckChan chan gw.DownlinkTXAck
	uplinkFrameChan   chan gw.UplinkFrame
	gatewayStatsChan  chan gw.GatewayStats
	udpSendChan       chan udpPacket

	wg             sync.WaitGroup
	conn           *net.UDPConn
	closed         bool
	gateways       gateways
	fakeRxTime     bool
	configurations []pfConfiguration
	skipCRCCheck   bool
}

// NewBackend creates a new backend.
func NewBackend(conf config.Config) (*Backend, error) {
	addr, err := net.ResolveUDPAddr("udp", conf.Backend.SemtechUDP.UDPBind)
	if err != nil {
		return nil, errors.Wrap(err, "resolve udp addr error")
	}

	log.WithField("addr", addr).Info("backend/semtechudp: starting gateway udp listener")
	conn, err := net.ListenUDP("udp", addr)
	if err != nil {
		return nil, errors.Wrap(err, "listen udp error")
	}

	b := &Backend{
		conn:              conn,
		downlinkTXAckChan: make(chan gw.DownlinkTXAck),
		uplinkFrameChan:   make(chan gw.UplinkFrame),
		gatewayStatsChan:  make(chan gw.GatewayStats),
		udpSendChan:       make(chan udpPacket),
		gateways: gateways{
			gateways:           make(map[lorawan.EUI64]gateway),
			subscribeEventChan: make(chan events.Subscribe),
		},
		fakeRxTime:   conf.Backend.SemtechUDP.FakeRxTime,
		skipCRCCheck: conf.Backend.SemtechUDP.SkipCRCCheck,
		tokenMap:     make(map[uint16][]byte),
	}

	for _, pfConf := range conf.Backend.SemtechUDP.Configuration {
		c := pfConfiguration{
			baseFile:       pfConf.BaseFile,
			outputFile:     pfConf.OutputFile,
			restartCommand: pfConf.RestartCommand,
		}
		if err := c.gatewayID.UnmarshalText([]byte(pfConf.GatewayID)); err != nil {
			return nil, errors.Wrap(err, "unmarshal gateway id error")
		}
		b.configurations = append(b.configurations, c)
	}

	go func() {
		for {
			log.Debug("backend/semtechudp: cleanup gateway registry")
			if err := b.gateways.cleanup(); err != nil {
				log.WithError(err).Error("backend/semtechudp: gateway registry cleanup failed")
			}
			time.Sleep(time.Minute)
		}
	}()

	go func() {
		b.wg.Add(1)
		err := b.readPackets()
		if !b.isClosed() {
			log.WithError(err).Error("backend/semtechudp: read udp packets error")
		}
		b.wg.Done()
	}()

	go func() {
		b.wg.Add(1)
		err := b.sendPackets()
		if !b.isClosed() {
			log.WithError(err).Error("backend/semtechudp: send udp packets error")
		}
		b.wg.Done()
	}()

	return b, nil
}

// Close closes the backend.
func (b *Backend) Close() error {
	b.Lock()<commit_msg>Fix race condition in semtechudp/backend (#162)<commit_after>package semtechudp

import (
	"crypto/rand"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"sync"
	"time"

	"github.com/pkg/errors"
	log "github.com/sirupsen/logrus"

	"github.com/brocaar/chirpstack-api/go/v3/gw"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/events"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/backend/semtechudp/packets"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/config"
	"github.com/brocaar/chirpstack-gateway-bridge/internal/filters"
	"github.com/brocaar/lorawan"
)

// udpPacket represents a raw UDP packet.
type udpPacket struct {
	addr *net.UDPAddr
	data []byte
}

type pfConfiguration struct {
	gatewayID      lorawan.EUI64
	baseFile       string
	outputFile     string
	restartCommand string
	currentVersion string
}

// Backend implements a Semtech packet-forwarder (UDP) gateway backend.
type Backend struct {
	sync.RWMutex

	// tokenMap stores the token to UUID mapping. This should take ~ 1MB of
	// memory. Optionaly this could be optimized by letting keys expire after
	// a given time.
	tokenMap map[uint16][]byte

	downlinkTXAckChan chan gw.DownlinkTXAck
	uplinkFrameChan   chan gw.UplinkFrame
	gatewayStatsChan  chan gw.GatewayStats
	udpSendChan       chan udpPacket

	wg             sync.WaitGroup
	conn           *net.UDPConn
	closed         bool
	gateways       gateways
	fakeRxTime     bool
	configurations []pfConfiguration
	skipCRCCheck   bool
}

// NewBackend creates a new backend.
func NewBackend(conf config.Config) (*Backend, error) {
	addr, err := net.ResolveUDPAddr("udp", conf.Backend.SemtechUDP.UDPBind)
	if err != nil {
		return nil, errors.Wrap(err, "resolve udp addr error")
	}

	log.WithField("addr", addr).Info("backend/semtechudp: starting gateway udp listener")
	conn, err := net.ListenUDP("udp", addr)
	if err != nil {
		return nil, errors.Wrap(err, "listen udp error")
	}

	b := &Backend{
		conn:              conn,
		downlinkTXAckChan: make(chan gw.DownlinkTXAck),
		uplinkFrameChan:   make(chan gw.UplinkFrame),
		gatewayStatsChan:  make(chan gw.GatewayStats),
		udpSendChan:       make(chan udpPacket),
		gateways: gateways{
			gateways:           make(map[lorawan.EUI64]gateway),
			subscribeEventChan: make(chan events.Subscribe),
		},
		fakeRxTime:   conf.Backend.SemtechUDP.FakeRxTime,
		skipCRCCheck: conf.Backend.SemtechUDP.SkipCRCCheck,
		tokenMap:     make(map[uint16][]byte),
	}

	for _, pfConf := range conf.Backend.SemtechUDP.Configuration {
		c := pfConfiguration{
			baseFile:       pfConf.BaseFile,
			outputFile:     pfConf.OutputFile,
			restartCommand: pfConf.RestartCommand,
		}
		if err := c.gatewayID.UnmarshalText([]byte(pfConf.GatewayID)); err != nil {
			return nil, errors.Wrap(err, "unmarshal gateway id error")
		}
		b.configurations = append(b.configurations, c)
	}

	go func() {
		for {
			log.Debug("backend/semtechudp: cleanup gateway registry")
			if err := b.gateways.cleanup(); err != nil {
				log.WithError(err).Error("backend/semtechudp: gateway registry cleanup failed")
			}
			time.Sleep(time.Minute)
		}
	}()

	// Add the waitgroups before the goroutines or a race occurs with closing
	b.wg.Add(2)
	go func() {
		err := b.readPackets()
		if !b.isClosed() {
			log.WithError(err).Error("backend/semtechudp: read udp packets error")
		}
		b.wg.Done()
	}()

	go func() {
		err := b.sendPackets()
		if !b.isClosed() {
			log.WithError(err).Error("backend/semtechudp: send udp packets error")
		}
		b.wg.Done()
	}()

	return b, nil
}

// Close closes the backend.
func (b *Backend) Close() error {
	b.Lock()
{'fraction_non_alphanumeric': 0.12121212121212122, 'fraction_numerical': 0.004170141784820684, 'mean_word_length': 3.6811971372804164, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3504309', 'n_tokens_mistral': 2808, 'n_tokens_neox': 2621, 'n_words': 553}