refactor(p2p): use buffered channels to communicate inside peer connection

Author: superp00t

faws/app/repository/notify.go

@@ -11,16 +11,18 @@ import (
 	"github.com/faws-vcs/faws/faws/app"
 	"github.com/faws-vcs/faws/faws/repo/cas"
 	"github.com/faws-vcs/faws/faws/repo/event"
+	"github.com/faws-vcs/faws/faws/repo/p2p/peernet"
 )
 
 var (
 	stages_text = map[event.Stage]string{
-		event.StagePullObjects: "Retrieve objects",
-		event.StagePullTags:    "Retrieve tags",
-		event.StageCacheFiles:  "Cache files",
-		event.StageCacheFile:   "Cache file",
-		event.StageWriteTree:   "Write tree",
-		event.StageCheckout:    "Checkout",
+		event.StagePullObjects:  "Retrieve objects",
+		event.StagePullTags:     "Retrieve tags",
+		event.StageCacheFiles:   "Cache files",
+		event.StageCacheFile:    "Cache file",
+		event.StageWriteTree:    "Write tree",
+		event.StageCheckout:     "Checkout",
+		event.StageServeObjects: "Distribute objects",
 	}
 
 	scrn activity_screen
@@ -54,6 +56,12 @@ type activity_screen struct {
 
 	connected_peers int
 
+	received_messages              int64
+	last_message_id                peernet.MessageID
+	object_uploads                 int64
+	duplicate_object_downloads     int64
+	duplicate_object_download_size uint64
+
 	verbose bool
 }
 
@@ -166,6 +174,14 @@ func notify(ev event.Notification, params *event.NotifyParams) {
 	case event.NotifyPeerDisconnected:
 		// app.Info("disconnected from", params.ID)
 		scrn.connected_peers--
+	case event.NotifyPeerNetMessage:
+		scrn.received_messages++
+		scrn.last_message_id = params.MessageID
+	case event.NotifyPeerObjectUpload:
+		scrn.object_uploads++
+	case event.NotifyPeerObjectDuplicateDownload:
+		scrn.duplicate_object_downloads++
+		scrn.duplicate_object_download_size += uint64(params.Count)
 	}
 	guard.Unlock()
 
@@ -230,6 +246,21 @@ func render_activity_screen(hud *console.Hud) {
 		return
 	}
 
+	// if scrn.received_messages > 0 {
+	// 	var received_messages_text console.Text
+	// 	received_messages_text.Stylesheet.Width = console.Width()
+	// 	received_messages_text.Add(fmt.Sprintf("%d messages received. last message id: %s", scrn.received_messages, scrn.last_message_id), 0, 0)
+	// 	hud.Line(&received_messages_text)
+	// }
+
+	if scrn.connected_peers > 0 {
+		var peers_text console.Text
+		peers_text.Stylesheet.Margin[console.Left] = 1
+		peers_text.Stylesheet.Width = console.Width()
+		peers_text.Add(fmt.Sprintf("%d peers connected", scrn.connected_peers), 0, 0)
+		hud.Line(&peers_text)
+	}
+
 	var progress_bar console.ProgressBar
 	progress_bar.Stylesheet.Sequence[console.PbCaseLeft] = console.Cell{'[', 0, 0}
 	progress_bar.Stylesheet.Sequence[console.PbCaseRight] = console.Cell{']', 0, 0}
@@ -249,18 +280,23 @@ func render_activity_screen(hud *console.Hud) {
 		progress_bar.Progress = float64(scrn.tags_received) / float64(scrn.tags_in_queue)
 		hud.Line(&progress_bar)
 	case event.StagePullObjects:
-		if scrn.connected_peers > 0 {
-			var peers_text console.Text
-			peers_text.Stylesheet.Margin[console.Left] = 1
-			peers_text.Stylesheet.Width = console.Width()
-			peers_text.Add(fmt.Sprintf("%d peers connected", scrn.connected_peers), 0, 0)
-			hud.Line(&peers_text)
+
+		if scrn.duplicate_object_downloads > 0 {
+			var duplicate_objects_text console.Text
+			duplicate_objects_text.Stylesheet.Margin[console.Left] = 1
+			duplicate_objects_text.Stylesheet.Width = console.Width()
+			duplicate_objects_text.Add(
+				fmt.Sprintf("%d duplicate objects downloaded (%s wasted)", scrn.duplicate_object_downloads, humanize.Bytes(scrn.duplicate_object_download_size)),
+				console.BrightYellow,
+				0,
+			)
+			hud.Line(&duplicate_objects_text)
 		}
 
 		var usage_text console.Text
 		usage_text.Stylesheet.Width = console.Width()
 		usage_text.Stylesheet.Margin[console.Left] = 1
-		usage_text.Add(fmt.Sprintf("%d/%d objects received, %s total", scrn.objects_received, scrn.objects_in_queue, humanize.Bytes(scrn.bytes_received)), 0, 0)
+		usage_text.Add(fmt.Sprintf("%d/%d objects processed, %s total", scrn.objects_received, scrn.objects_in_queue, humanize.Bytes(scrn.bytes_received)), 0, 0)
 		hud.Line(&usage_text)
 
 		// if scrn.objects_received > 0 {
@@ -305,5 +341,10 @@ func render_activity_screen(hud *console.Hud) {
 
 		hud.Line(&file_name_text)
 		hud.Line(&progress_text, &progress_bar)
+	case event.StageServeObjects:
+		var object_upload_count_text console.Text
+		object_upload_count_text.Stylesheet.Width = console.Width()
+		object_upload_count_text.Add(fmt.Sprintf("%d objects uploaded", scrn.object_uploads), 0, 0)
+		hud.Line(&object_upload_count_text)
 	}
 }

faws/app/repository/pull.go

@@ -13,6 +13,7 @@ type PullParams struct {
 	// If len(refs) == 0, download all tags from the origin
 	Tags    bool
 	Verbose bool
+	Quiet   bool
 }
 
 // Pull is the implementation of the command "faws pull"
@@ -23,6 +24,7 @@ func Pull(params *PullParams) {
 		TrackerURL = params.TrackerURL
 	}
 
+	quiet = params.Quiet
 	app.Open()
 	defer func() {
 		app.Close()

faws/app/repository/seed.go

@@ -11,6 +11,7 @@ type SeedParams struct {
 	Sign       string
 	TrackerURL string
 	TopicURI   string
+	Quiet      bool
 }
 
 func Seed(params *SeedParams) {
@@ -23,6 +24,8 @@ func Seed(params *SeedParams) {
 		TrackerURL = params.TrackerURL
 	}
 
+	quiet = params.Quiet
+
 	app.Open()
 	defer func() {
 		app.Close()

faws/app/repository/util.go

@@ -7,6 +7,7 @@ import (
 	"github.com/faws-vcs/faws/faws/app"
 	"github.com/faws-vcs/faws/faws/app/identities"
 	"github.com/faws-vcs/faws/faws/repo"
+	"github.com/faws-vcs/faws/faws/repo/event"
 	"github.com/faws-vcs/faws/faws/repo/p2p/tracker"
 )
 
@@ -15,16 +16,26 @@ var Repo repo.Repository
 
 var TrackerURL = tracker.DefaultURL
 
+var quiet bool
+
 // Open opens the repository located at directory
 func Open(directory string) (err error) {
+	notify_func := notify
+
+	if quiet {
+		notify_func = func(ev event.Notification, params *event.NotifyParams) {}
+	}
+
 	err = Repo.Open(directory,
 		repo.WithTrust(identities.NewRingTrust(app.Configuration.Ring())),
-		repo.WithNotify(notify),
+		repo.WithNotify(notify_func),
 		repo.WithTracker(TrackerURL),
 	)
 
-	console.RenderFunc(render_activity_screen)
-	console.SwapInterval(time.Second / 3)
+	if !quiet {
+		console.RenderFunc(render_activity_screen)
+		console.SwapInterval(time.Second / 3)
+	}
 	return
 }
 

faws/cmd/pull/pull.go

@@ -21,6 +21,7 @@ func init() {
 	flag := pull_cmd.Flags()
 	flag.BoolP("tag", "t", false, "pull the named tags instead of objects. If no tags are named, all tags from the origin will get pulled")
 	flag.BoolP("verbose", "v", false, "display extra information")
+	flag.BoolP("quiet", "q", false, "shut up the interactive Hud")
 	root.RootCmd.AddCommand(&pull_cmd)
 }
 
@@ -61,5 +62,10 @@ func run_pull_cmd(cmd *cobra.Command, args []string) {
 		app.Fatal(err)
 		return
 	}
+	params.Quiet, err = flag.GetBool("quiet")
+	if err != nil {
+		app.Fatal(err)
+		return
+	}
 	repository.Pull(&params)
 }

faws/cmd/seed/seed.go

@@ -20,6 +20,7 @@ var seed_cmd = cobra.Command{
 func init() {
 	flag := seed_cmd.Flags()
 	flag.StringP("sign", "s", "", "use a signing identity to identify yourself with the P2P network")
+	flag.BoolP("quiet", "q", false, "shut up the interactive Hud")
 	root.RootCmd.AddCommand(&seed_cmd)
 }
 
@@ -47,5 +48,10 @@ func run_seed_cmd(cmd *cobra.Command, args []string) {
 		app.Fatal(err)
 		return
 	}
+	params.Quiet, err = flag.GetBool("quiet")
+	if err != nil {
+		app.Fatal(err)
+		return
+	}
 	repository.Seed(&params)
 }

faws/repo/event/event.go

@@ -3,6 +3,7 @@ package event
 import (
 	"github.com/faws-vcs/faws/faws/identity"
 	"github.com/faws-vcs/faws/faws/repo/cas"
+	"github.com/faws-vcs/faws/faws/repo/p2p/peernet"
 )
 
 // A Notification signifies different types of repository events
@@ -29,6 +30,9 @@ const (
 	// p2p
 	NotifyPeerConnected
 	NotifyPeerDisconnected
+	NotifyPeerNetMessage
+	NotifyPeerObjectUpload
+	NotifyPeerObjectDuplicateDownload
 )
 
 // A Stage represents a phase of operations within the repository, typically one that can take quite a long time.
@@ -45,6 +49,7 @@ const (
 	StagePullTags
 	StagePullObjects
 	StageCheckout
+	StageServeObjects
 )
 
 // NotifyParams are extra information parameters shared along with the Notification
@@ -67,6 +72,8 @@ type NotifyParams struct {
 	Child bool
 	//
 	ID identity.ID
+	//
+	MessageID peernet.MessageID
 }
 
 // A NotifyFunc can be supplied to repo.Repository.Open to get notifications about the repository's actions

faws/repo/p2p/agent.go

@@ -20,6 +20,7 @@ type Agent struct {
 }
 
 func (agent *Agent) Init(options ...Option) (err error) {
+	agent.options.requests_per_second = 128
 	agent.options.tracker_url = tracker.DefaultURL
 	agent.options.notify = func(n event.Notification, params *event.NotifyParams) {}
 

faws/repo/p2p/agent_options.go

@@ -17,8 +17,9 @@ type agent_options struct {
 	// URL of the tracker server
 	tracker_url string
 	// If true, forces the use of a TURN server
-	use_turn bool
-	notify   event.NotifyFunc
+	use_turn            bool
+	notify              event.NotifyFunc
+	requests_per_second int64
 }
 
 type Option func(*agent_options)

faws/repo/p2p/job.go

@@ -105,7 +105,7 @@ func (agent *Agent) Seed(topic tracker.Topic) (job Job, err error) {
 	}
 
 	seed_job_ := new(seed_job)
-	seed_job_.init()
+	seed_job_.init(subscription)
 
 	subscription.set_current_job(seed_job_)
 	job = seed_job_