s416496/musique
1
0
Fork 0
Code Issues 3 Pull Requests 1 Projects Releases Wiki Activity

musique<->server integration

Browse Source
This commit is contained in:
Robert Bendun 2022-12-16 16:34:52 +01:00
parent ffc65e0f06
commit c23eae555c
16 changed files with 310 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@ -5,4 +5,4 @@ RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
RUN apt update && apt upgrade -y && apt install -y build-essential software-properties-common zip unzip
RUN add-apt-repository ppa:ubuntu-toolchain-r/test
RUN apt install -y gcc-11 g++-11 libasound2-dev
RUN apt install -y gcc-11 g++-11 libasound2-dev golang

51
musique/config.cc
View File

@ -3,6 +3,13 @@
#include <musique/config.hh>
#include <musique/errors.hh>
#include <musique/platform.hh>
#include <iostream>
#ifdef MUSIQUE_WINDOWS
#include <shlobj.h>
#include <knownfolders.h>
#include <cstring>
#endif
// FIXME Move trim to some header
extern void trim(std::string_view &s);
@ -39,10 +46,26 @@ config::Sections config::from_file(std::string const& path)
return sections;
}
void config::to_file(std::string const& path, Sections const& sections)
{
std::ofstream out(path, std::ios_base::trunc | std::ios_base::out);
if (!out.is_open()) {
std::cout << "Failed to save configuration at " << path << std::endl;
return;
}
for (auto const& [section, kv] : sections) {
out << '[' << section << "]\n";
for (auto const& [key, val] : kv) {
out << key << " = " << val << '\n';
}
}
}
std::string config::location()
{
if constexpr (Current_Platform == Platform::Linux) {
#ifdef MUSIQUE_LINUX
if (auto config_dir = getenv("XDG_CONFIG_HOME")) {
return config_dir + std::string("/musique.conf");
} else if (auto home_dir = getenv("HOME")) {
@ -50,7 +73,29 @@ std::string config::location()
} else {
unimplemented("Neither XDG_CONFIG_HOME nor HOME enviroment variable are defined");
}
} else {
#endif
#ifdef MUSIQUE_WINDOWS
wchar_t *resultPath = nullptr;
SHGetKnownFolderPath(FOLDERID_RoamingAppData, 0, nullptr, &resultPath);
auto const len = std::strlen((char*)resultPath);
if (!resultPath && len < 3) {
auto drive = getenv("HOMEDRIVE");
auto path = getenv("HOMEPATH");
ensure(drive && path, "Windows failed to provide HOMEDRIVE & HOMEPATH variables");
return std::string(drive) + std::string(path) + "\\Documents\\musique.conf";
}
return std::string((char*)resultPath, len) + "\\musique.conf";
#endif
#ifdef MUSIQUE_DARWIN
if (auto home_dir = getenv("HOME")) {
return std::string(home_dir) + "/Library/Application Support/musique.conf";
} else {
unimplemented("HOME enviroment variable is not defined");
}
#endif
unimplemented();
}
}

1
musique/config.hh
View File

@ -12,6 +12,7 @@ namespace config
using Sections = std::unordered_map<std::string, Key_Value>;
Sections from_file(std::string const& path);
void to_file(std::string const& path, Sections const& sections);
std::string location();
}

6
musique/interpreter/builtin_functions.cc
View File

@ -1115,12 +1115,18 @@ static Result<Value> builtin_start(Interpreter &interpreter, std::span<Ast> args
{
Value ret{};
auto begin = std::chrono::steady_clock::now();
ServerBeginProtocol();
for (auto const& ast : args) {
ret = Try(interpreter.eval((Ast)ast));
}
auto end = std::chrono::steady_clock::now();
std::cout << "Start took " << std::chrono::duration_cast<std::chrono::duration<float>>(end - begin) << "s" << std::endl;
return ret;
}

35
musique/main.cc
View File

@ -106,6 +106,8 @@ void print_repl_help()
":!<command> - allows for execution of any shell command\n"
":clear - clears screen\n"
":load <file> - loads file into Musique session\n"
":discover - tries to discover new remotes\n"
":remotes - list all known remotes\n"
;
}
@ -207,6 +209,16 @@ struct Runner
std::cout << std::endl;
return {};
});
Env::global->force_define("timeout", +[](Interpreter&, std::vector<Value> args) -> Result<Value> {
if (auto a = match<Number>(args); a) {
auto [timeout] = *a;
auto gotimeout = GoInt64((timeout * Number(1000)).floor().as_int());
SetTimeout(gotimeout);
return Value{};
}
unimplemented();
});
}
Runner(Runner const&) = delete;
@ -346,7 +358,15 @@ static Result<bool> handle_repl_session_commands(std::string_view input, Runner
ListKnownRemotes();
return std::nullopt;
}
}
},
Command {
"discover",
+[](Runner&, std::optional<std::string_view>) -> std::optional<Error> {
Discover();
ListKnownRemotes();
return std::nullopt;
}
},
};
if (input.starts_with('!')) {
@ -458,6 +478,8 @@ static std::optional<Error> Main(std::span<char const*> args)
std::exit(1);
}
bool save_new_config = false;
// TODO Write configuration
// TODO Nicer configuration interface (maybe paths?)
auto config = config::from_file(config::location());
@ -466,6 +488,11 @@ static std::optional<Error> Main(std::span<char const*> args)
} else {
std::cout << "Please enter your nick: ";
std::getline(std::cin, nick);
auto nick_view = std::string_view(nick);
trim(nick_view); // FIXME Trim in place
nick = nick_view;
config["net"]["nick"] = nick;
save_new_config = true;
}
if (config.contains("net") && config["net"].contains("port")) {
@ -474,6 +501,12 @@ static std::optional<Error> Main(std::span<char const*> args)
std::from_chars(port_str.data(), port_str.data() + port_str.size(), port);
} else {
port = 8081;
config["net"]["port"] = std::to_string(port);
save_new_config = true;
}
if (save_new_config) {
config::to_file(config::location(), config);
}
Runner runner{output_port};

3
musique/platform.hh
View File

@ -10,10 +10,13 @@ enum class Platform
#if defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__)
static constexpr Platform Current_Platform = Platform::Windows;
#define MUSIQUE_WINDOWS
#elif __APPLE__
static constexpr Platform Current_Platform = Platform::Darwin;
#define MUSIQUE_DARWIN
#else
static constexpr Platform Current_Platform = Platform::Linux;
#define MUSIQUE_LINUX
#endif
#endif // MUSIQUE_PLATFORM_HH

2
scripts/release
View File

@ -19,7 +19,9 @@ mkdir -p "$Target"
if [[ "$(docker images -q "$Image")" == "" ]]; then
docker build -t "$Image" .
fi
sudo rm -rf bin/
# make os=linux CC=gcc CXX=g++ >/dev/null
docker run -it --rm -v "$(pwd):/musique" -w /musique "$Image" make os=linux CC=gcc-11 CXX=g++-11 >/dev/null
cp bin/musique "$Target"/musique-x86_64-linux

2
scripts/windows.mk
View File

@ -1,7 +1,7 @@
CC=x86_64-w64-mingw32-cc
CXX=x86_64-w64-mingw32-c++
CPPFLAGS:=$(CPPFLAGS) -D__WINDOWS_MM__
LDLIBS:=-lwinmm $(LDLIBS) -static-libgcc -static-libstdc++ -static
LDLIBS:=-lwinmm -luuid $(LDLIBS) -static-libgcc -static-libstdc++ -static
Target=musique.exe
GOOS=windows
GOARCH=amd64

23
server/go.sum Normal file
View File

@ -0,0 +1,23 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/k0kubun/go-ansi v0.0.0-20180517002512-3bf9e2903213/go.mod h1:vNUNkEQ1e29fT/6vq2aBdFsgNPmy8qMdSay1npru+Sw=
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
github.com/mattn/go-runewidth v0.0.14 h1:+xnbZSEeDbOIg5/mE6JF0w6n9duR1l3/WmbinWVwUuU=
github.com/mattn/go-runewidth v0.0.14/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db h1:62I3jR2EmQ4l5rM/4FEfDWcRD+abF5XlKShorW5LRoQ=
github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db/go.mod h1:l0dey0ia/Uv7NcFFVbCLtqEBQbrT4OCwCSKTEv6enCw=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/rivo/uniseg v0.4.3 h1:utMvzDsuh3suAEnhH0RdHmoPbU648o6CvXxTx4SBMOw=
github.com/rivo/uniseg v0.4.3/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
github.com/schollz/progressbar v1.0.0 h1:gbyFReLHDkZo8mxy/dLWMr+Mpb1MokGJ1FqCiqacjZM=
github.com/schollz/progressbar v1.0.0/go.mod h1:/l9I7PC3L3erOuz54ghIRKUEFcosiWfLvJv+Eq26UMs=
github.com/schollz/progressbar/v3 v3.12.2 h1:yLqqqpQNMxGxHY8uEshRihaHWwa0rf0yb7/Zrpgq2C0=
github.com/schollz/progressbar/v3 v3.12.2/go.mod h1:HFJYIYQQJX32UJdyoigUl19xoV6aMwZt6iX/C30RWfg=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.3.0 h1:w8ZOecv6NaNa/zC8944JTU3vz4u6Lagfk4RPQxv92NQ=
golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.3.0 h1:qoo4akIqOcDME5bhc/NgxUdovd6BSS2uMsVjB56q1xI=
golang.org/x/term v0.3.0/go.mod h1:q750SLmJuPmVoN1blW3UFBPREJfb1KmY3vwxfr+nFDA=

206
server/main.go
View File

@ -1,14 +1,13 @@
package main
import (
"bufio"
"errors"
"flag"
"fmt"
"log"
"musique/server/proto"
"musique/server/scan"
"musique/server/router"
"musique/server/scan"
"net"
"os"
"strings"
@ -22,65 +21,39 @@ func scanError(scanResult []string, conn net.Conn) {
}
}
type timeExchange struct {
type TimeExchange struct {
before, after, remote int64
}
type client struct {
timeExchange
id int
addr string
}
func remotef(host, command, format string, args ...interface{}) (int, error) {
connection, err := net.Dial("tcp", host)
if err != nil {
return 0, fmt.Errorf("remotef: establishing connection: %v", err)
}
defer connection.Close()
connection.SetDeadline(time.Now().Add(1 * time.Second))
fmt.Fprintln(connection, command)
if len(format) > 0 {
parsedCount, err := fmt.Fscanf(connection, format, args...)
if err != nil {
return 0, fmt.Errorf("remotef: parsing: %v", err)
}
return parsedCount, nil
}
return 0, nil
}
func (e *timeExchange) estimateFor(host string) bool {
func (e *TimeExchange) estimateFor(host string) bool {
e.before = time.Now().UnixMilli()
parsedCount, err := remotef(host, "time", "%d\n", &e.remote)
var response proto.TimeResponse
if err := proto.Command(host, proto.Time(), &response); err != nil {
log.Printf("failed to send time command: %v\n", err)
return false
}
e.after = time.Now().UnixMilli()
if err != nil {
log.Printf("estimateFor: %v\n", err)
return false
}
if parsedCount != 1 {
log.Printf("berkeley: expected to parse number, instead parsed %d items\n", parsedCount)
return false
}
return true
}
func timesync(hosts []string) []client {
func timesync() {
wg := sync.WaitGroup{}
wg.Add(len(hosts))
wg.Add(len(remotes))
type response struct {
TimeExchange
key string
}
// Gather time from each host
responses := make(chan client, len(hosts))
for id, host := range hosts {
id, host := id, host
responses := make(chan response, len(remotes))
for key, remote := range remotes {
key, remote := key, remote
go func() {
defer wg.Done()
exchange := timeExchange{}
if exchange.estimateFor(host) {
responses <- client{exchange, id, host}
exchange := TimeExchange{}
if exchange.estimateFor(remote.Address) {
responses <- response{exchange, key}
}
}()
}
@ -88,17 +61,13 @@ func timesync(hosts []string) []client {
wg.Wait()
close(responses)
clients := make([]client, 0, len(hosts))
for client := range responses {
clients = append(clients, client)
for response := range responses {
remote := remotes[response.key]
remote.TimeExchange = response.TimeExchange
}
fmt.Printf("Successfully gathered %d clients\n", len(clients))
return clients
}
const maxReactionTime = 300
var maxReactionTime = int64(1_000)
func isThisMyAddress(address string) bool {
addrs, err := net.InterfaceAddrs()
@ -116,18 +85,24 @@ func isThisMyAddress(address string) bool {
return false
}
func notifyAll(clients []client) <-chan time.Time {
func notifyAll() <-chan time.Time {
wg := sync.WaitGroup{}
wg.Add(len(clients))
startDeadline := time.After(maxReactionTime * time.Millisecond)
wg.Add(len(remotes))
startDeadline := time.After(time.Duration(maxReactionTime) * time.Millisecond)
for _, client := range clients {
client := client
for _, remote := range remotes {
remote := remote
go func() {
startTime := maxReactionTime - (client.after-client.before)/2
_, err := remotef(client.addr, fmt.Sprintf("start %d", startTime), "")
startTime := maxReactionTime - (remote.after-remote.before)/2
var response proto.StartResponse
err := proto.CommandTimeout(
remote.Address,
proto.Start(startTime),
&response,
time.Duration(startTime)*time.Millisecond,
)
if err != nil {
log.Printf("failed to notify %s: %v\n", client.addr, err)
log.Printf("failed to notify %s: %v\n", remote.Address, err)
}
wg.Done()
}()
@ -136,6 +111,10 @@ func notifyAll(clients []client) <-chan time.Time {
return startDeadline
}
var (
pinger chan struct{}
)
func registerRoutes(r *router.Router) {
r.Add("handshake", func(incoming net.Conn, request proto.Request) interface{} {
var response proto.HandshakeResponse
@ -160,14 +139,31 @@ func registerRoutes(r *router.Router) {
return synchronizeHosts(request.HostsResponse)
})
r.Add("synchronize-hosts-with-remotes", func(incoming net.Conn, request proto.Request) interface{} {
synchronizeHostsWithRemotes()
return nil
})
r.Add("time", func(incoming net.Conn, request proto.Request) interface{} {
return proto.TimeResponse{
Time: time.Now().UnixMilli(),
}
})
pinger = make(chan struct{}, 12)
r.Add("start", func(incoming net.Conn, request proto.Request) interface{} {
go func() {
time.Sleep(time.Duration(request.StartTime) * time.Millisecond)
pinger <- struct{}{}
}()
return proto.StartResponse{
Succeded: true,
}
})
}
type Remote struct {
TimeExchange
Address string
Nick string
Version string
@ -177,7 +173,7 @@ var (
baseIP string = ""
nick string
port int = 8888
remotes map[string]Remote
remotes map[string]*Remote
)
func synchronizeHosts(incoming proto.HostsResponse) (response proto.HostsResponse) {
@ -187,7 +183,7 @@ func synchronizeHosts(incoming proto.HostsResponse) (response proto.HostsRespons
// Additionaly build set of all hosts that remote knows
for _, incomingHost := range incoming.Hosts {
if _, ok := remotes[incomingHost.Address]; !ok && !isThisMyAddress(incomingHost.Address) {
remotes[incomingHost.Address] = Remote{
remotes[incomingHost.Address] = &Remote{
Address: incomingHost.Address,
Nick: incomingHost.Nick,
Version: incomingHost.Version,
@ -285,10 +281,10 @@ func registerRemotes() error {
hosts := scan.TCPHosts(networks, []uint16{8081, 8082, 8083, 8084})
remotes = make(map[string]Remote)
remotes = make(map[string]*Remote)
for host := range hosts {
if !isThisMyAddress(host.Address) {
remotes[host.Address] = Remote{
remotes[host.Address] = &Remote{
Address: host.Address,
Nick: host.Nick,
Version: host.Version,
@ -338,75 +334,3 @@ func main() {
for range exit {
}
}
func main2() {
l, err := net.Listen("tcp", ":8081")
if err != nil {
log.Fatal(err)
}
defer l.Close()
for {
conn, err := l.Accept()
if err != nil {
log.Fatal(err)
}
go func(c net.Conn) {
s := bufio.NewScanner(c)
scanResult := []string{
"10.100.5.112:8081",
"10.100.5.44:8081",
}
var clients []client
for s.Scan() {
resp := s.Text()
log.Println(resp)
if resp == "scan" {
conn.Write([]byte("Scanning...\n"))
scanResult = nil // scan()
conn.Write([]byte("Scanning done!\n"))
fmt.Println(len(scanResult))
continue
}
if resp == "time" {
fmt.Fprintln(conn, time.Now().UnixMilli())
continue
}
if resp == "hosts" {
scanError(scanResult, conn)
for _, host := range scanResult {
conn.Write([]byte(host + "\n"))
fmt.Println("CONNECTED")
}
continue
}
if resp == "showtime" {
cTime := showTime()
conn.Write([]byte(cTime.String() + "\n"))
continue
}
if resp == "timesync" {
clients = timesync(scanResult)
continue
}
if strings.HasPrefix(resp, "start") {
startTimeString := strings.TrimSpace(resp[len("start"):])
startTime := int64(0)
fmt.Sscanf(startTimeString, "%d", &startTime)
time.Sleep(time.Duration(startTime) * time.Millisecond)
log.Println("Started #start")
continue
}
if resp == "notify" {
<-notifyAll(clients)
log.Println("Started #notify")
continue
}
if resp == "quit" {
c.Close()
os.Exit(0)
}
}
c.Close()
}(conn)
}
}

34
server/musique-bridge.go
View File

@ -11,12 +11,18 @@ import (
//export ServerInit
func ServerInit(inputNick string, inputPort int) {
logFile, err := os.OpenFile("musique.log", os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o640)
if err != nil {
log.Fatalln(err)
}
log.SetOutput(logFile)
nick = inputNick
port = inputPort
r := router.Router{}
registerRoutes(&r)
_, err := r.Run(baseIP, uint16(port))
_, err = r.Run(baseIP, uint16(port))
if err != nil {
log.Fatalln(err)
}
@ -24,10 +30,32 @@ func ServerInit(inputNick string, inputPort int) {
if err := registerRemotes(); err != nil {
log.Fatalln(err)
}
timesync()
}
//export SetTimeout
func SetTimeout(timeout int64) {
maxReactionTime = timeout
}
//export ServerBeginProtocol
func ServerBeginProtocol() {
self := notifyAll()
select {
case <-self:
case <-pinger:
}
}
//export Discover
func Discover() {
if len(remotes) == 0 {
if err := registerRemotes(); err != nil {
log.Println("discover:", err)
}
}
synchronizeHostsWithRemotes()
}
//export ListKnownRemotes
@ -38,10 +66,10 @@ func ListKnownRemotes() {
list := []nickAddr{}
for _, remote := range remotes {
list = append(list, nickAddr { remote.Nick, remote.Address})
list = append(list, nickAddr{remote.Nick, remote.Address})
}
sort.Slice(list, func (i, j int) bool {
sort.Slice(list, func(i, j int) bool {
if list[i].nick == list[j].nick {
return list[i].addr < list[j].addr
}

1
server/proto/basic.go
View File

@ -6,4 +6,5 @@ type Request struct {
Version string
Type string
HostsResponse
StartTime int64
}

12
server/proto/start.go Normal file
View File

@ -0,0 +1,12 @@
package proto
type StartResponse struct {
Succeded bool
}
func Start(startTime int64) (req Request) {
req.Type = "start"
req.Version = Version
req.StartTime = startTime
return
}

11
server/proto/time.go Normal file
View File

@ -0,0 +1,11 @@
package proto
type TimeResponse struct {
Time int64
}
func Time() (req Request) {
req.Type = "time"
req.Version = Version
return
}

67
server/router/router.go Normal file
View File

@ -0,0 +1,67 @@
package router
import (
"encoding/json"
"musique/server/proto"
"net"
"log"
"fmt"
)
type Route func(incoming net.Conn, request proto.Request) interface{}
type Router struct {
routes map[string]Route
port uint16
baseIP string
}
func (router *Router) Add(name string, route Route) {
if router.routes == nil {
router.routes = make(map[string]Route)
}
router.routes[name] = route
}
func (router *Router) Run(ip string, port uint16) (<-chan struct{}, error) {
router.port = port
router.baseIP = ip
listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", router.baseIP, router.port))
if err != nil {
return nil, err
}
exit := make(chan struct{})
go func() {
defer listener.Close()
defer close(exit)
for {
incoming, err := listener.Accept()
if err != nil {
log.Println("failed to accept connection:", err)
}
go router.handleIncoming(incoming)
}
}()
return exit, nil
}
func (router *Router) handleIncoming(incoming net.Conn) {
defer incoming.Close()
request := proto.Request{}
json.NewDecoder(incoming).Decode(&request)
log.Printf("%s: %+v\n", incoming.RemoteAddr(), request)
if handler, ok := router.routes[request.Type]; ok {
if response := handler(incoming, request); response != nil {
json.NewEncoder(incoming).Encode(response)
}
} else {
log.Printf("unrecongized route: %v\n", request.Type)
}
}

6
server/scan/scanner.go
View File

@ -2,6 +2,7 @@ package scan
import (
"fmt"
"log"
"musique/server/proto"
"net"
"sync"
@ -58,10 +59,11 @@ type Response struct {
// TCPHosts returns all TCP hosts that are in given networks on one of given ports
func TCPHosts(networks []Network, ports []uint16) <-chan Response {
ips := make(chan Response, 32)
ips := make(chan Response, 256)
log.Printf("tcphosts: %+v\n", networks)
wg := sync.WaitGroup{}
for _, network := range networks {
ip := net.ParseIP(network.FirstAddress)
for i := 0; i < network.MaxHostsCount; i++ {