<commit_before>// Copyright (C) 2017 Jennal(<EMAIL>). All rights reserved.
//
// Licensed under the MIT License (the "License"); you may not use this file except
// in compliance with the License. You may obtain a copy of the License at
//
// http://opensource.org/licenses/MIT
//
// Unless required by applicable law or agreed to in writing, software distributed
// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
// CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.

package heartbeat

import (
	"fmt"
	"sync"
	"time"

	"github.com/jennal/goplay/consts"
	"github.com/jennal/goplay/log"
	"github.com/jennal/goplay/pkg"
	"github.com/jennal/goplay/session"
)

const (
	TIMEOUT     time.Duration = 3000 * time.Millisecond
	INTERNAL    time.Duration = consts.HeartBeatRate * time.Millisecond
	MAX_TIMEOUT int           = 3
)

type HeartBeatProcessor struct {
	mapMutex     sync.Mutex
	timeoutMutex sync.Mutex

	manager *HeartBeatManager
	sess    *session.Session
	times   map[pkg.PackageIDType]time.Time

	lastPing     time.Duration
	avgPing      time.Duration
	minPing      time.Duration
	maxPing      time.Duration
	sendCount    int64
	recvCount    int64
	timeoutCount int

	stopSignal bool
}

func NewHeartBeatProcessor(manager *HeartBeatManager) *HeartBeatProcessor {
	result := &HeartBeatProcessor{
		manager: manager,
		times:   make(map[pkg.PackageIDType]time.Time),
	}
	go result.checkTimeOut()

	return result
}

func (self *HeartBeatProcessor) Stop() {
	self.stopSignal = true
}

func (self *HeartBeatProcessor) pushTime(id pkg.PackageIDType, t time.Time) {
	self.mapMutex.Lock()
	defer self.mapMutex.Unlock()

	self.times[id] = t
}

func (self *HeartBeatProcessor) popTime(id pkg.PackageIDType) *time.Time {
	self.mapMutex.Lock()
	defer self.mapMutex.Unlock()

	item, ok := self.times[id]
	if !ok {
		return nil
	}
	delete(self.times, id)

	return &item
}

func (self *HeartBeatProcessor) checkTimeOut() {
	for {
		if self.stopSignal {
			return
		}

		// fmt.Printf("[%v] Check Timeout\n", time.Now())
		ids := []pkg.PackageIDType{}
		now := time.Now()

		for id, val := range self.times {
			if now.Sub(val) > TIMEOUT {
				// fmt.Printf("\t => Timeout: %v => %v, id = %v\n", now, val, id)
				ids = append(ids, id)
			}
		}

		for _, id := range ids {
			self.popTime(id)
			if !self.incTimeOut() {
				return
			}<commit_msg>add lock to heartbeat processor checkTimeOut<commit_after>// Copyright (C) 2017 Jennal(<EMAIL>). All rights reserved.
//
// Licensed under the MIT License (the "License"); you may not use this file except
// in compliance with the License. You may obtain a copy of the License at
//
// http://opensource.org/licenses/MIT
//
// Unless required by applicable law or agreed to in writing, software distributed
// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
// CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.

package heartbeat

import (
	"fmt"
	"sync"
	"time"

	"github.com/jennal/goplay/consts"
	"github.com/jennal/goplay/log"
	"github.com/jennal/goplay/pkg"
	"github.com/jennal/goplay/session"
)

const (
	TIMEOUT     time.Duration = 3000 * time.Millisecond
	INTERNAL    time.Duration = consts.HeartBeatRate * time.Millisecond
	MAX_TIMEOUT int           = 3
)

type HeartBeatProcessor struct {
	mapMutex     sync.Mutex
	timeoutMutex sync.Mutex

	manager *HeartBeatManager
	sess    *session.Session
	times   map[pkg.PackageIDType]time.Time

	lastPing     time.Duration
	avgPing      time.Duration
	minPing      time.Duration
	maxPing      time.Duration
	sendCount    int64
	recvCount    int64
	timeoutCount int

	stopSignal bool
}

func NewHeartBeatProcessor(manager *HeartBeatManager) *HeartBeatProcessor {
	result := &HeartBeatProcessor{
		manager: manager,
		times:   make(map[pkg.PackageIDType]time.Time),
	}
	go result.checkTimeOut()

	return result
}

func (self *HeartBeatProcessor) Stop() {
	self.stopSignal = true
}

func (self *HeartBeatProcessor) pushTime(id pkg.PackageIDType, t time.Time) {
	self.mapMutex.Lock()
	defer self.mapMutex.Unlock()

	self.times[id] = t
}

func (self *HeartBeatProcessor) popTime(id pkg.PackageIDType) *time.Time {
	self.mapMutex.Lock()
	defer self.mapMutex.Unlock()

	item, ok := self.times[id]
	if !ok {
		return nil
	}
	delete(self.times, id)

	return &item
}

func (self *HeartBeatProcessor) checkTimeOut() {
	for {
		if self.stopSignal {
			return
		}

		// fmt.Printf("[%v] Check Timeout\n", time.Now())
		ids := []pkg.PackageIDType{}
		now := time.Now()

		self.mapMutex.Lock()
		for id, val := range self.times {
			if now.Sub(val) > TIMEOUT {
				// fmt.Printf("\t => Timeout: %v => %v, id = %v\n", now, val, id)
				ids = append(ids, id)
			}
		}
		self.mapMutex.Unlock()

		for _, id := range ids {
			self.popTime(id)
			if !self.incTimeOut() {
				return
			}
{'fraction_non_alphanumeric': 0.1156530408773679, 'fraction_numerical': 0.00518444666001994, 'mean_word_length': 3.486583184257603, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '2635290', 'n_tokens_mistral': 1819, 'n_tokens_neox': 1674, 'n_words': 469}