- {{template "views/partials/navbar"}}
+ {{template "views/partials/navbar" .}}
diff --git a/core/p2p/node.go b/core/p2p/node.go
new file mode 100644
index 00000000..1d5356e6
--- /dev/null
+++ b/core/p2p/node.go
@@ -0,0 +1,50 @@
+package p2p
+
+import (
+ "sync"
+ "time"
+)
+
+const defaultServicesID = "services_localai"
+const FederatedID = "federated"
+
+type NodeData struct {
+ Name string
+ ID string
+ TunnelAddress string
+ LastSeen time.Time
+}
+
+func (d NodeData) IsOnline() bool {
+ now := time.Now()
+ // if the node was seen in the last 40 seconds, it's online
+ return now.Sub(d.LastSeen) < 40*time.Second
+}
+
+var mu sync.Mutex
+var nodes = map[string]map[string]NodeData{}
+
+func GetAvailableNodes(serviceID string) []NodeData {
+ if serviceID == "" {
+ serviceID = defaultServicesID
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ var availableNodes = []NodeData{}
+ for _, v := range nodes[serviceID] {
+ availableNodes = append(availableNodes, v)
+ }
+ return availableNodes
+}
+
+func AddNode(serviceID string, node NodeData) {
+ if serviceID == "" {
+ serviceID = defaultServicesID
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if nodes[serviceID] == nil {
+ nodes[serviceID] = map[string]NodeData{}
+ }
+ nodes[serviceID][node.ID] = node
+}
diff --git a/core/p2p/p2p.go b/core/p2p/p2p.go
index 67196c97..79e8051e 100644
--- a/core/p2p/p2p.go
+++ b/core/p2p/p2p.go
@@ -10,19 +10,18 @@ import (
"io"
"net"
"os"
- "strings"
+ "sync"
"time"
+ "github.com/ipfs/go-log"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/mudler/LocalAI/pkg/utils"
+ "github.com/mudler/edgevpn/pkg/config"
"github.com/mudler/edgevpn/pkg/node"
"github.com/mudler/edgevpn/pkg/protocol"
+ "github.com/mudler/edgevpn/pkg/services"
"github.com/mudler/edgevpn/pkg/types"
"github.com/phayes/freeport"
-
- "github.com/ipfs/go-log"
- "github.com/mudler/edgevpn/pkg/config"
- "github.com/mudler/edgevpn/pkg/services"
zlog "github.com/rs/zerolog/log"
"github.com/mudler/edgevpn/pkg/logger"
@@ -34,6 +33,15 @@ func GenerateToken() string {
return newData.Base64()
}
+func IsP2PEnabled() bool {
+ return true
+}
+
+func nodeID(s string) string {
+ hostname, _ := os.Hostname()
+ return fmt.Sprintf("%s-%s", hostname, s)
+}
+
func allocateLocalService(ctx context.Context, node *node.Node, listenAddr, service string) error {
zlog.Info().Msgf("Allocating service '%s' on: %s", service, listenAddr)
@@ -53,16 +61,16 @@ func allocateLocalService(ctx context.Context, node *node.Node, listenAddr, serv
10*time.Second,
func() {
// Retrieve current ID for ip in the blockchain
- _, found := ledger.GetKey(protocol.UsersLedgerKey, node.Host().ID().String())
+ //_, found := ledger.GetKey(protocol.UsersLedgerKey, node.Host().ID().String())
// If mismatch, update the blockchain
- if !found {
- updatedMap := map[string]interface{}{}
- updatedMap[node.Host().ID().String()] = &types.User{
- PeerID: node.Host().ID().String(),
- Timestamp: time.Now().String(),
- }
- ledger.Add(protocol.UsersLedgerKey, updatedMap)
+ //if !found {
+ updatedMap := map[string]interface{}{}
+ updatedMap[node.Host().ID().String()] = &types.User{
+ PeerID: node.Host().ID().String(),
+ Timestamp: time.Now().String(),
}
+ ledger.Add(protocol.UsersLedgerKey, updatedMap)
+ // }
},
)
@@ -80,7 +88,6 @@ func allocateLocalService(ctx context.Context, node *node.Node, listenAddr, serv
continue
}
- // ll.Info("New connection from", l.Addr().String())
// Handle connections in a new goroutine, forwarding to the p2p service
go func() {
// Retrieve current ID for ip in the blockchain
@@ -137,24 +144,30 @@ func copyStream(closer chan struct{}, dst io.Writer, src io.Reader) {
// This is the main of the server (which keeps the env variable updated)
// This starts a goroutine that keeps LLAMACPP_GRPC_SERVERS updated with the discovered services
-func LLamaCPPRPCServerDiscoverer(ctx context.Context, token string) error {
- tunnels, err := discoveryTunnels(ctx, token)
+func ServiceDiscoverer(ctx context.Context, n *node.Node, token, servicesID string, discoveryFunc func()) error {
+ if servicesID == "" {
+ servicesID = defaultServicesID
+ }
+ tunnels, err := discoveryTunnels(ctx, n, token, servicesID)
if err != nil {
return err
}
-
+ // TODO: discoveryTunnels should return all the nodes that are available?
+ // In this way we updated availableNodes here instead of appending
+ // e.g. we have a LastSeen field in NodeData that is updated in discoveryTunnels
+ // each time the node is seen
+ // In this case the below function should be idempotent and just keep track of the nodes
go func() {
- totalTunnels := []string{}
for {
select {
case <-ctx.Done():
zlog.Error().Msg("Discoverer stopped")
return
case tunnel := <-tunnels:
-
- totalTunnels = append(totalTunnels, tunnel)
- os.Setenv("LLAMACPP_GRPC_SERVERS", strings.Join(totalTunnels, ","))
- zlog.Debug().Msgf("setting LLAMACPP_GRPC_SERVERS to %s", strings.Join(totalTunnels, ","))
+ AddNode(servicesID, tunnel)
+ if discoveryFunc != nil {
+ discoveryFunc()
+ }
}
}
}()
@@ -162,19 +175,10 @@ func LLamaCPPRPCServerDiscoverer(ctx context.Context, token string) error {
return nil
}
-func discoveryTunnels(ctx context.Context, token string) (chan string, error) {
- tunnels := make(chan string)
+func discoveryTunnels(ctx context.Context, n *node.Node, token, servicesID string) (chan NodeData, error) {
+ tunnels := make(chan NodeData)
- nodeOpts, err := newNodeOpts(token)
- if err != nil {
- return nil, err
- }
-
- n, err := node.New(nodeOpts...)
- if err != nil {
- return nil, fmt.Errorf("creating a new node: %w", err)
- }
- err = n.Start(ctx)
+ err := n.Start(ctx)
if err != nil {
return nil, fmt.Errorf("creating a new node: %w", err)
}
@@ -184,8 +188,14 @@ func discoveryTunnels(ctx context.Context, token string) (chan string, error) {
}
// get new services, allocate and return to the channel
+
+ // TODO:
+ // a function ensureServices that:
+ // - starts a service if not started, if the worker is Online
+ // - checks that workers are Online, if not cancel the context of allocateLocalService
+ // - discoveryTunnels should return all the nodes and addresses associated with it
+ // - the caller should take now care of the fact that we are always returning fresh informations
go func() {
- emitted := map[string]bool{}
for {
select {
case <-ctx.Done():
@@ -195,20 +205,20 @@ func discoveryTunnels(ctx context.Context, token string) (chan string, error) {
time.Sleep(5 * time.Second)
zlog.Debug().Msg("Searching for workers")
- data := ledger.LastBlock().Storage["services_localai"]
- for k := range data {
+ data := ledger.LastBlock().Storage[servicesID]
+ for k, v := range data {
zlog.Info().Msgf("Found worker %s", k)
- if _, found := emitted[k]; !found {
- emitted[k] = true
- //discoveredPeers <- k
- port, err := freeport.GetFreePort()
- if err != nil {
- fmt.Print(err)
- }
- tunnelAddress := fmt.Sprintf("127.0.0.1:%d", port)
- go allocateLocalService(ctx, n, tunnelAddress, k)
- tunnels <- tunnelAddress
+ nd := &NodeData{}
+ if err := v.Unmarshal(nd); err != nil {
+ zlog.Error().Msg("cannot unmarshal node data")
+ continue
}
+ ensureService(ctx, n, nd, k)
+ muservice.Lock()
+ if _, ok := service[nd.Name]; ok {
+ tunnels <- service[nd.Name].NodeData
+ }
+ muservice.Unlock()
}
}
}
@@ -217,8 +227,60 @@ func discoveryTunnels(ctx context.Context, token string) (chan string, error) {
return tunnels, err
}
+type nodeServiceData struct {
+ NodeData NodeData
+ CancelFunc context.CancelFunc
+}
+
+var service = map[string]nodeServiceData{}
+var muservice sync.Mutex
+
+func ensureService(ctx context.Context, n *node.Node, nd *NodeData, sserv string) {
+ muservice.Lock()
+ defer muservice.Unlock()
+ if ndService, found := service[nd.Name]; !found {
+ if !nd.IsOnline() {
+ // if node is offline and not present, do nothing
+ return
+ }
+ newCtxm, cancel := context.WithCancel(ctx)
+ // Start the service
+ port, err := freeport.GetFreePort()
+ if err != nil {
+ fmt.Print(err)
+ }
+ tunnelAddress := fmt.Sprintf("127.0.0.1:%d", port)
+ nd.TunnelAddress = tunnelAddress
+ service[nd.Name] = nodeServiceData{
+ NodeData: *nd,
+ CancelFunc: cancel,
+ }
+ go allocateLocalService(newCtxm, n, tunnelAddress, sserv)
+ zlog.Debug().Msgf("Starting service %s on %s", sserv, tunnelAddress)
+ } else {
+ // Check if the service is still alive
+ // if not cancel the context
+ if !nd.IsOnline() && !ndService.NodeData.IsOnline() {
+ ndService.CancelFunc()
+ delete(service, nd.Name)
+ zlog.Info().Msgf("Node %s is offline, deleting", nd.ID)
+ } else if nd.IsOnline() {
+ // update last seen inside service
+ nd.TunnelAddress = ndService.NodeData.TunnelAddress
+ service[nd.Name] = nodeServiceData{
+ NodeData: *nd,
+ CancelFunc: ndService.CancelFunc,
+ }
+ zlog.Debug().Msgf("Node %s is still online", nd.ID)
+ }
+ }
+}
+
// This is the P2P worker main
-func BindLLamaCPPWorker(ctx context.Context, host, port, token string) error {
+func ExposeService(ctx context.Context, host, port, token, servicesID string) error {
+ if servicesID == "" {
+ servicesID = defaultServicesID
+ }
llger := logger.New(log.LevelFatal)
nodeOpts, err := newNodeOpts(token)
@@ -248,22 +310,40 @@ func BindLLamaCPPWorker(ctx context.Context, host, port, token string) error {
ledger.Announce(
ctx,
- 10*time.Second,
+ 20*time.Second,
func() {
// Retrieve current ID for ip in the blockchain
- _, found := ledger.GetKey("services_localai", name)
+ //_, found := ledger.GetKey("services_localai", name)
// If mismatch, update the blockchain
- if !found {
- updatedMap := map[string]interface{}{}
- updatedMap[name] = "p2p"
- ledger.Add("services_localai", updatedMap)
+ //if !found {
+ updatedMap := map[string]interface{}{}
+ updatedMap[name] = &NodeData{
+ Name: name,
+ LastSeen: time.Now(),
+ ID: nodeID(name),
}
+ ledger.Add(servicesID, updatedMap)
+ // }
},
)
return err
}
+func NewNode(token string) (*node.Node, error) {
+ nodeOpts, err := newNodeOpts(token)
+ if err != nil {
+ return nil, err
+ }
+
+ n, err := node.New(nodeOpts...)
+ if err != nil {
+ return nil, fmt.Errorf("creating a new node: %w", err)
+ }
+
+ return n, nil
+}
+
func newNodeOpts(token string) ([]node.Option, error) {
llger := logger.New(log.LevelFatal)
defaultInterval := 10 * time.Second
diff --git a/core/p2p/p2p_disabled.go b/core/p2p/p2p_disabled.go
index 59314aa6..340a4fb4 100644
--- a/core/p2p/p2p_disabled.go
+++ b/core/p2p/p2p_disabled.go
@@ -6,16 +6,26 @@ package p2p
import (
"context"
"fmt"
+
+ "github.com/mudler/edgevpn/pkg/node"
)
func GenerateToken() string {
return "not implemented"
}
-func LLamaCPPRPCServerDiscoverer(ctx context.Context, token string) error {
+func ServiceDiscoverer(ctx context.Context, node *node.Node, token, servicesID string, fn func()) error {
return fmt.Errorf("not implemented")
}
-func BindLLamaCPPWorker(ctx context.Context, host, port, token string) error {
+func ExposeService(ctx context.Context, host, port, token, servicesID string) error {
return fmt.Errorf("not implemented")
}
+
+func IsP2PEnabled() bool {
+ return false
+}
+
+func NewNode(token string) (*node.Node, error) {
+ return nil, fmt.Errorf("not implemented")
+}
diff --git a/docs/static/install.sh b/docs/static/install.sh
index 4efdc757..61be710b 100644
--- a/docs/static/install.sh
+++ b/docs/static/install.sh
@@ -76,6 +76,8 @@ DOCKER_INSTALL=${DOCKER_INSTALL:-$docker_found}
USE_AIO=${USE_AIO:-false}
API_KEY=${API_KEY:-}
CORE_IMAGES=${CORE_IMAGES:-false}
+P2P_TOKEN=${P2P_TOKEN:-}
+WORKER=${WORKER:-false}
# nprocs -1
if available nproc; then
procs=$(nproc)
@@ -132,7 +134,14 @@ configure_systemd() {
info "Adding current user to local-ai group..."
$SUDO usermod -a -G local-ai $(whoami)
-
+ STARTCOMMAND="run"
+ if [ "$WORKER" = true ]; then
+ if [ -n "$P2P_TOKEN" ]; then
+ STARTCOMMAND="worker p2p-llama-cpp-rpc"
+ else
+ STARTCOMMAND="worker llama-cpp-rpc"
+ fi
+ fi
info "Creating local-ai systemd service..."
cat </dev/null
[Unit]
@@ -140,7 +149,7 @@ Description=LocalAI Service
After=network-online.target
[Service]
-ExecStart=$BINDIR/local-ai run
+ExecStart=$BINDIR/local-ai $STARTCOMMAND
User=local-ai
Group=local-ai
Restart=always
@@ -159,6 +168,11 @@ EOF
$SUDO echo "THREADS=$THREADS" | $SUDO tee -a /etc/localai.env >/dev/null
$SUDO echo "MODELS_PATH=$MODELS_PATH" | $SUDO tee -a /etc/localai.env >/dev/null
+ if [ -n "$P2P_TOKEN" ]; then
+ $SUDO echo "LOCALAI_P2P_TOKEN=$P2P_TOKEN" | $SUDO tee -a /etc/localai.env >/dev/null
+ $SUDO echo "LOCALAI_P2P=true" | $SUDO tee -a /etc/localai.env >/dev/null
+ fi
+
SYSTEMCTL_RUNNING="$(systemctl is-system-running || true)"
case $SYSTEMCTL_RUNNING in
running|degraded)
@@ -407,6 +421,19 @@ install_docker() {
# exit 0
fi
+ STARTCOMMAND="run"
+ if [ "$WORKER" = true ]; then
+ if [ -n "$P2P_TOKEN" ]; then
+ STARTCOMMAND="worker p2p-llama-cpp-rpc"
+ else
+ STARTCOMMAND="worker llama-cpp-rpc"
+ fi
+ fi
+ envs=""
+ if [ -n "$P2P_TOKEN" ]; then
+ envs="-e LOCALAI_P2P_TOKEN=$P2P_TOKEN -e LOCALAI_P2P=true"
+ fi
+
IMAGE_TAG=
if [ "$HAS_CUDA" ]; then
IMAGE_TAG=${VERSION}-cublas-cuda12-ffmpeg
@@ -430,7 +457,8 @@ install_docker() {
--restart=always \
-e API_KEY=$API_KEY \
-e THREADS=$THREADS \
- -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG
+ $envs \
+ -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG $STARTCOMMAND
elif [ "$HAS_AMD" ]; then
IMAGE_TAG=${VERSION}-hipblas-ffmpeg
# CORE
@@ -448,7 +476,8 @@ install_docker() {
--restart=always \
-e API_KEY=$API_KEY \
-e THREADS=$THREADS \
- -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG
+ $envs \
+ -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG $STARTCOMMAND
elif [ "$HAS_INTEL" ]; then
IMAGE_TAG=${VERSION}-sycl-f32-ffmpeg
# CORE
@@ -465,7 +494,8 @@ install_docker() {
--restart=always \
-e API_KEY=$API_KEY \
-e THREADS=$THREADS \
- -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG
+ $envs \
+ -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG $STARTCOMMAND
else
IMAGE_TAG=${VERSION}-ffmpeg
# CORE
@@ -481,7 +511,8 @@ install_docker() {
-e MODELS_PATH=/models \
-e API_KEY=$API_KEY \
-e THREADS=$THREADS \
- -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG
+ $envs \
+ -d -p $PORT:8080 --name local-ai localai/localai:$IMAGE_TAG $STARTCOMMAND
fi
install_success
diff --git a/embedded/webui_static.yaml b/embedded/webui_static.yaml
index d9d1f064..c65d64ee 100644
--- a/embedded/webui_static.yaml
+++ b/embedded/webui_static.yaml
@@ -16,6 +16,9 @@
- filename: "tw-elements.css"
url: "https://cdn.jsdelivr.net/npm/tw-elements/css/tw-elements.min.css"
sha: "72746af5326d6eb3647f504efa81b5e0f50ed486f37cc8262a4169781ad310d3"
+- filename: "tw-elements.js"
+ url: "https://cdn.jsdelivr.net/npm/tw-elements/js/tw-elements.umd.min.js"
+ sha: "2985706362e92360b65c8697cc32490bb9c0a5df9cd9b7251a97c1c5a661a40a"
- filename: "tailwindcss.js"
url: "https://cdn.tailwindcss.com/3.3.0"
sha: "dbff048aa4581e6eae7f1cb2c641f72655ea833b3bb82923c4a59822e11ca594"