waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

add some synchronization magic for ptypos updates

Browse Source
This commit is contained in:
sawka
2023-03-28 00:24:37 -07:00
parent e674333621
commit dde547351c
4 changed files with 71 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/pcloud/pcloud.go
+29 -11
View File
@@ -24,7 +24,7 @@ import (
const PCloudEndpoint = "https://api.getprompt.dev/central"
const PCloudEndpointVarName = "PCLOUD_ENDPOINT"
const APIVersion = 1
const MaxPtyUpdateSize = (128 * 1024) + 1
const MaxPtyUpdateSize = (128 * 1024)
const MaxUpdatesPerReq = 10
const MaxUpdateWriterErrors = 3
const PCloudDefaultTimeout = 5 * time.Second
@@ -291,11 +291,19 @@ func makeWebShareUpdate(ctx context.Context, update *sstore.ScreenUpdateType) (*
if err != nil {
return nil, fmt.Errorf("error getting ptypos: %v", err)
}
realOffset, data, err := sstore.ReadPtyOutFile(ctx, update.ScreenId, cmdId, ptyPos, MaxPtyUpdateSize)
sstore.SetWebScreenPtyPosDelIntent(update.ScreenId, update.LineId)
realOffset, data, err := sstore.ReadPtyOutFile(ctx, update.ScreenId, cmdId, ptyPos, MaxPtyUpdateSize+1)
if err != nil {
return nil, fmt.Errorf("error getting ptydata: %v", err)
}
rtn.PtyData = &WebSharePtyData{PtyPos: realOffset, Data: data}
if len(data) == 0 {
return nil, nil
}
if len(data) > MaxPtyUpdateSize {
rtn.PtyData = &WebSharePtyData{PtyPos: realOffset, Data: data[0:MaxPtyUpdateSize], Eof: false}
} else {
rtn.PtyData = &WebSharePtyData{PtyPos: realOffset, Data: data, Eof: true}
}
default:
return nil, fmt.Errorf("unsupported update type (pcloud/makeWebScreenUpdate): %s\n", update.UpdateType)
@@ -306,7 +314,7 @@ func makeWebShareUpdate(ctx context.Context, update *sstore.ScreenUpdateType) (*
func finalizeWebScreenUpdate(ctx context.Context, webUpdate *WebShareUpdateType) error {
switch webUpdate.UpdateType {
case sstore.UpdateType_PtyPos:
dataEof := len(webUpdate.PtyData.Data) < MaxPtyUpdateSize
dataEof := webUpdate.PtyData.Eof
newPos := webUpdate.PtyData.PtyPos + int64(len(webUpdate.PtyData.Data))
if dataEof {
err := sstore.RemoveScreenUpdate(ctx, webUpdate.UpdateId)
@@ -318,6 +326,10 @@ func finalizeWebScreenUpdate(ctx context.Context, webUpdate *WebShareUpdateType)
if err != nil {
return err
}
err = sstore.MaybeRemovePtyPosUpdate(ctx, webUpdate.ScreenId, webUpdate.LineId, webUpdate.UpdateId)
if err != nil {
return err
}
default:
err := sstore.RemoveScreenUpdate(ctx, webUpdate.UpdateId)
@@ -338,21 +350,27 @@ func DoWebScreenUpdates(authInfo AuthInfo, updateArr []*sstore.ScreenUpdateType)
var webUpdates []*WebShareUpdateType
for _, update := range updateArr {
webUpdate, err := makeWebShareUpdate(context.Background(), update)
if err != nil {
// log error, remove update, and continue
log.Printf("[pcloud] error create web-share update updateid:%d: %v", update.UpdateId, err)
err = sstore.RemoveScreenUpdate(context.Background(), update.UpdateId)
if err != nil || webUpdate == nil {
// log error (if there is one), remove update, and continue
if err != nil {
log.Printf("[pcloud] error create web-share update updateid:%d: %v", update.UpdateId, err)
}
if update.UpdateType == sstore.UpdateType_PtyPos {
err = sstore.MaybeRemovePtyPosUpdate(context.Background(), update.ScreenId, update.LineId, update.UpdateId)
} else {
err = sstore.RemoveScreenUpdate(context.Background(), update.UpdateId)
}
if err != nil {
// ignore this error too (although this is really problematic, there is nothing to do)
log.Printf("[pcloud] error removing screen update updateid:%d: %v", update.UpdateId, err)
}
continue
}
if webUpdate == nil {
continue
}
webUpdates = append(webUpdates, webUpdate)
}
if len(webUpdates) == 0 {
return nil
}
ctx, cancelFn := context.WithTimeout(context.Background(), PCloudDefaultTimeout)
defer cancelFn()
req, err := makeAuthPostReq(ctx, WebShareUpdateUrl, authInfo, webUpdates)
pkg/pcloud/pclouddata.go
+1
View File
@@ -156,4 +156,5 @@ func webCmdFromCmd(lineId string, cmd *sstore.CmdType) (*WebShareCmdType, error)
type WebSharePtyData struct {
PtyPos int64 `json:"ptypos,omitempty"`
Data []byte `json:"data,omitempty"`
Eof bool `json:"-"` // internal use
}
pkg/sstore/dbops.go
+40 -2
View File
@@ -24,6 +24,8 @@ const DefaultMaxHistoryItems = 1000
var updateWriterCVar = sync.NewCond(&sync.Mutex{})
var updateWriterMoreData = false
var WebScreenPtyPosLock = &sync.Mutex{}
var WebScreenPtyPosDelIntent = make(map[string]bool) // map[screenid + ":" + lineid] -> bool
type SingleConnDBGetter struct {
SingleConnLock *sync.Mutex
@@ -2400,12 +2402,16 @@ func ScreenWebShareStart(ctx context.Context, screenId string, shareOpts ScreenW
if shareMode != ShareModeLocal {
return fmt.Errorf("screen cannot be shared, invalid current share mode %q (must be local)", shareMode)
}
nowTs := time.Now().UnixMilli()
query = `UPDATE screen SET sharemode = ?, webshareopts = ? WHERE screenid = ?`
tx.Exec(query, ShareModeWeb, quickJson(shareOpts), screenId)
insertScreenUpdate(tx, screenId, UpdateType_ScreenNew)
query = `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets)
SELECT screenid, lineid, ?, ? FROM line WHERE screenid = ? ORDER BY linenum`
tx.Exec(query, UpdateType_LineNew, time.Now().UnixMilli(), screenId)
tx.Exec(query, UpdateType_LineNew, nowTs, screenId)
query = `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets)
SELECT c.screenid, l.lineid, ?, ? FROM cmd c, line l WHERE c.screenid = ? AND l.cmdid = c.cmdid`
tx.Exec(query, UpdateType_PtyPos, nowTs, screenId)
NotifyUpdateWriter()
return nil
})
@@ -2425,6 +2431,8 @@ func ScreenWebShareStop(ctx context.Context, screenId string) error {
tx.Exec(query, ShareModeLocal, "null", screenId)
query = `DELETE FROM screenupdate WHERE screenid = ?`
tx.Exec(query, screenId)
query = `DELETE FROM webptypos WHERE screenid = ?`
tx.Exec(query, screenId)
insertScreenUpdate(tx, screenId, UpdateType_ScreenDel)
return nil
})
@@ -2494,13 +2502,31 @@ func RemoveScreenUpdate(ctx context.Context, updateId int64) error {
})
}
func InsertPtyPosUpdate(ctx context.Context, screenId string, cmdId string) error {
func SetWebScreenPtyPosDelIntent(screenId string, lineId string) {
WebScreenPtyPosLock.Lock()
defer WebScreenPtyPosLock.Unlock()
WebScreenPtyPosDelIntent[screenId+":"+lineId] = true
}
func ClearWebScreenPtyPosDelIntent(screenId string, lineId string) bool {
WebScreenPtyPosLock.Lock()
defer WebScreenPtyPosLock.Unlock()
rtn := WebScreenPtyPosDelIntent[screenId+":"+lineId]
delete(WebScreenPtyPosDelIntent, screenId+":"+lineId)
return rtn
}
func MaybeInsertPtyPosUpdate(ctx context.Context, screenId string, cmdId string) error {
return WithTx(ctx, func(tx *TxWrap) error {
if !isWebShare(tx, screenId) {
return nil
}
query := `SELECT lineid FROM line WHERE screenid = ? AND cmdid = ?`
lineId := tx.GetString(query, screenId, cmdId)
if lineId == "" {
return fmt.Errorf("invalid ptypos update, no lineid found for %s/%s", screenId, cmdId)
}
ClearWebScreenPtyPosDelIntent(screenId, lineId) // clear delete intention because we have a new update
query = `SELECT updateid FROM screenupdate WHERE screenid = ? AND lineid = ? AND updatetype = ?`
if !tx.Exists(query, screenId, lineId, UpdateType_PtyPos) {
query := `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) VALUES (?, ?, ?, ?)`
@@ -2511,6 +2537,18 @@ func InsertPtyPosUpdate(ctx context.Context, screenId string, cmdId string) erro
})
}
func MaybeRemovePtyPosUpdate(ctx context.Context, screenId string, lineId string, updateId int64) error {
return WithTx(ctx, func(tx *TxWrap) error {
intent := ClearWebScreenPtyPosDelIntent(screenId, lineId) // check for intention before deleting
if !intent {
return nil
}
query := `DELETE FROM screenupdate WHERE updateid = ?`
tx.Exec(query, updateId)
return nil
})
}
func GetWebPtyPos(ctx context.Context, screenId string, lineId string) (int64, error) {
return WithTxRtn(ctx, func(tx *TxWrap) (int64, error) {
query := `SELECT ptypos FROM webptypos WHERE screenid = ? AND lineid = ?`
pkg/sstore/fileops.go
+1 -1
View File
@@ -61,7 +61,7 @@ func AppendToCmdPtyBlob(ctx context.Context, screenId string, cmdId string, data
PtyData64: data64,
PtyDataLen: int64(len(data)),
}
err = InsertPtyPosUpdate(ctx, screenId, cmdId)
err = MaybeInsertPtyPosUpdate(ctx, screenId, cmdId)
if err != nil {
// just log
log.Printf("error inserting ptypos update %s/%s: %v\n", screenId, cmdId, err)