Poller split custom number #23
Replies: 2 comments 20 replies
-
|
Hi, I'll check tomorrow what can be done. I was planning to make splitting more configurable/extendable. At the moment there is logic that you can use maybe as a workaround. If you add specific query parameters to
If you want to know how it exactly works the place to look is Lines 184 to 193 in 77dd4be |
Beta Was this translation helpful? Give feedback.
-
Holes also dont work. List of registers are. |
Beta Was this translation helpful? Give feedback.
-
|
I assume that you device does not handle multiple connections well. It is probably quite dumb eth->serial device that can not do multiple parallel request because behind ethernet there is actually single serial connection. by default poller splits fields to requests and tries to poll them in parallel. I actually have code to force same ip+port request to be parallel. For that you need actually have custom client that uses mutexes. I used following code to limit serial connections type serialClientPool struct {
mu sync.RWMutex
clientPool map[string]*sharedClient
logger *zerolog.Logger
}
func (scp *serialClientPool) onClose(address string) {
scp.mu.Lock()
defer scp.mu.Unlock()
delete(scp.clientPool, address)
}
func (scp *serialClientPool) newSerialClient(batchProtocol mb.ProtocolType, address string) (poller.Client, error) {
scp.mu.Lock()
defer scp.mu.Unlock()
if batchProtocol != mb.ProtocolRTU {
return nil, fmt.Errorf("invalid batch protocol: %v", batchProtocol)
}
// address can have params `/dev/ttyS0:9600?invalid_addr=70`
if i := strings.IndexByte(address, '?'); i != -1 {
address = address[:i]
}
client, ok := scp.clientPool[address]
if ok {
scp.logger.Info().
Str("address", address).
Msg("modbus serial client found in pool")
return client, nil
}
deviceBaud := defaultSerialBaud
devicePath, rawBaud, ok := strings.Cut(address, ":")
if ok {
tmpBaud, err := strconv.Atoi(rawBaud)
if err != nil {
return nil, fmt.Errorf("device baud could not be converted to int")
}
deviceBaud = tmpBaud
}
port, err := serial.OpenPort(&serial.Config{
Name: devicePath,
Baud: cmp.Or(deviceBaud, defaultSerialBaud),
ReadTimeout: 1 * time.Second,
Size: 8,
})
if err != nil {
return nil, err
}
client = &sharedClient{
mu: sync.Mutex{},
client: mb.NewSerialClient(port),
address: address,
onClose: scp.onClose,
}
scp.clientPool[address] = client
scp.logger.Info().
Str("address", address).
Msg("new modbus serial client created")
return client, nil
}
type sharedClient struct {
mu sync.Mutex
client *mb.SerialClient
address string
onClose func(address string)
}
func (c *sharedClient) Do(ctx context.Context, req packet.Request) (packet.Response, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return nil, mb.ErrClientNotConnected
}
return c.client.Do(ctx, req)
}
func (c *sharedClient) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return nil
}
err := c.client.Close()
c.client = nil
c.onClose(c.address)
return err
}and poller is created as following: scp := serialClientPool{
mu: sync.RWMutex{},
clientPool: map[string]*sharedClient{},
logger: s.logger,
}
pollerConfig := poller.Config{
Logger: logger,
ConnectFunc: func(ctx context.Context, batchProtocol mb.ProtocolType, address string) (poller.Client, error) {
if strings.HasPrefix(address, "/dev/") {
return scp.newSerialClient(batchProtocol, address)
}
return poller.DefaultConnectClient(ctx, batchProtocol, address)
},
}
p := poller.NewPollerWithConfig(batches, pollerConfig)You could try to adapt to your needs. I'll look if I have time tomorrow to add something "use single connection" logic to the library itself. |
Beta Was this translation helpful? Give feedback.
-
|
I understand that you think about "dumb eth->serial device".
This convertor connected to Datacom or Mebay Genset controllers. |
Beta Was this translation helpful? Give feedback.
-
|
have you checked if that client is using single connection or perhaps slower/bigger interval for requests? At the moment poller creates separate goroutine and network connection for each splitted request. Serial devices can not handle parallel requests and usually what you see is timeout error and things like that. In your logs I see that sometimes some request succeed but other fail - I assume this is because too many requests are being done in parallel - as "serial" device name says - requests need to be done in serially. Datacom device may have ethernet support but this does not means that somewhere in background there is actually serial (uart/rs232/rs485) connection to engine ECU. |
Beta Was this translation helpful? Give feedback.
-
|
I didn’t check the client. |
Beta Was this translation helpful? Give feedback.
-
|
This will use single connection per address This code creates special shared/single connection function for Used as: config := poller.Config{
ConnectFunc: NewSingleConnectionPerAddressClientFunc(),
}
p := poller.NewPollerWithConfig(batches, config)type SingleConnectionPerAddress struct {
mu sync.Mutex
clientPool map[string]*sharedClient
}
func NewSingleConnectionPerAddressClientFunc() func(ctx context.Context, protocol modbus.ProtocolType, address string) (poller.Client, error) {
pool := &SingleConnectionPerAddress{
clientPool: map[string]*sharedClient{},
}
return pool.NewClientFunc
}
func (scp *SingleConnectionPerAddress) NewClientFunc(ctx context.Context, protocol modbus.ProtocolType, address string) (poller.Client, error) {
scp.mu.Lock()
defer scp.mu.Unlock()
// address could have params `tcp://192.168.1.2:502??invalid_addr=70`
if i := strings.IndexByte(address, '?'); i != -1 {
address = address[:i]
}
client, ok := scp.clientPool[address]
if ok {
return client, nil
}
orgClient, err := poller.DefaultConnectClient(ctx, protocol, address)
if err != nil {
return nil, err
}
client = &sharedClient{
mu: sync.Mutex{},
client: orgClient,
address: address,
onClose: scp.onClose,
}
scp.clientPool[address] = client
return client, nil
}
func (scp *SingleConnectionPerAddress) onClose(address string) {
scp.mu.Lock()
defer scp.mu.Unlock()
delete(scp.clientPool, address)
}
type sharedClient struct {
mu sync.Mutex
client poller.Client
address string
onClose func(address string)
}
func (c *sharedClient) Do(ctx context.Context, req packet.Request) (packet.Response, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return nil, modbus.ErrClientNotConnected
}
return c.client.Do(ctx, req)
}
func (c *sharedClient) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return nil
}
err := c.client.Close()
c.client = nil
c.onClose(c.address)
return err
} |
Beta Was this translation helpful? Give feedback.
-
|
Yes, this library will extract field values from response with modbus.BuilderResponse.ExtractFields method. In readme, for low level things this is that line For poller, complete example would be (single connection per server address included): package main
import (
"context"
"github.com/aldas/go-modbus-client"
"github.com/aldas/go-modbus-client/packet"
"github.com/aldas/go-modbus-client/poller"
"log/slog"
"os"
"os/signal"
"strings"
"sync"
"time"
)
func main() {
b := modbus.NewRequestBuilderWithConfig(modbus.BuilderDefaults{
ServerAddress: "tcp://172.25.28.29:8899?max_quantity_per_request=16",
FunctionCode: packet.FunctionReadHoldingRegisters,
UnitID: 1,
Protocol: modbus.ProtocolTCP,
Interval: modbus.Duration(5 * time.Second),
})
type Register struct {
Name string
Type modbus.FieldType
Address uint16
}
regs := []Register{
{"Main_l1_voltage", modbus.FieldTypeInt32, 10240},
{"Main_l2_voltage", modbus.FieldTypeInt32, 10242},
{"Main_l3_voltage", modbus.FieldTypeInt32, 10244},
{"Genset_l1_voltage", modbus.FieldTypeInt32, 10246},
{"Genset_l2_voltage", modbus.FieldTypeInt32, 10248},
{"Genset_l3_voltage", modbus.FieldTypeInt32, 10250},
{"Genset_l1_current", modbus.FieldTypeInt32, 10270},
{"Genset_l2_current", modbus.FieldTypeInt32, 10272},
{"Genset_l3_current", modbus.FieldTypeInt32, 10274},
{"Genset_total_current", modbus.FieldTypeInt32, 10278},
{"Genset_l1_active_power", modbus.FieldTypeInt32, 10286},
{"Genset_l2_active_power", modbus.FieldTypeInt32, 10288},
{"Genset_l3_active_power", modbus.FieldTypeInt32, 10290},
{"Genset_total_active_power", modbus.FieldTypeInt32, 10294},
{"Genset_l1_apparent_power", modbus.FieldTypeInt32, 10318},
{"Genset_l2_apparent_power", modbus.FieldTypeInt32, 10320},
{"Genset_l3_apparent_power", modbus.FieldTypeInt32, 10322},
{"Genset_total_apparent_power", modbus.FieldTypeInt32, 10326},
{"Main_frequency", modbus.FieldTypeUint16, 10338},
{"Genset_frequency", modbus.FieldTypeUint16, 10339},
{"Charge_input_voltage", modbus.FieldTypeUint16, 10340},
{"Battery_voltage", modbus.FieldTypeUint16, 10341},
{"Oil_presure", modbus.FieldTypeUint16, 10361},
{"Engine_temp", modbus.FieldTypeUint16, 10362},
{"Engine_rpm", modbus.FieldTypeUint16, 10376},
{"Unit_oper_status", modbus.FieldTypeUint16, 10604},
{"Unit_mode", modbus.FieldTypeUint16, 10605},
{"Count_genset_cranks", modbus.FieldTypeInt32, 10618},
{"Count_engine_hours_run", modbus.FieldTypeInt32, 10622},
}
for _, reg := range regs {
b.AddField(modbus.Field{
Name: reg.Name,
Address: reg.Address,
Type: reg.Type,
})
}
batches, _ := b.Split()
for i, batch := range batches {
slog.Info("Batch",
"index", i,
"StartAddress", batch.StartAddress,
"Quantity", batch.Quantity,
)
for j, field := range batch.Fields {
slog.Info("--- field",
"field_index", j,
"Name", field.Name,
"Address", field.Address,
)
}
}
doPolling(batches)
}
func doPolling(batches []modbus.BuilderRequest) {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()
config := poller.Config{
ConnectFunc: poller.NewSingleConnectionPerAddressClientFunc(poller.DefaultConnectClient),
}
p := poller.NewPollerWithConfig(batches, config)
go func() {
for {
select {
case result := <-p.ResultChan:
slog.Info("polled values", "values", result)
case <-ctx.Done():
return
}
}
}()
go func() {
tickerHealth := time.NewTicker(60 * time.Second)
defer tickerHealth.Stop()
for {
select {
case <-tickerHealth.C:
for _, s := range p.BatchStatistics() {
b := batches[s.BatchIndex]
slog.Info(
"batch health check",
"index", s.BatchIndex,
"address", s.ServerAddress,
"function_code", s.FunctionCode,
"start_address", b.StartAddress,
"quantity", b.Quantity,
"request_interval", b.RequestInterval.String(),
"is_polling", s.IsPolling,
"start_count", s.StartCount,
"request_ok_count", s.RequestOKCount,
"request_err_count", s.RequestErrCount,
"request_modbus_err_count", s.RequestModbusErrCount,
"send_skip_count", s.SendSkipCount,
)
}
case <-ctx.Done():
return
}
}
}()
if err := p.Poll(ctx); err != nil {
slog.Error("polling ended with failure", "err", err)
return
}
} |
Beta Was this translation helpful? Give feedback.
-
|
You will be definetely better of with larger request most of the time. For example if you latency (round-trip) from your machine to modbus server is 1ms. Then making 2 request takes at least 2ms but having larger quantity by adding couple bytes for register that are between wanted addreses will not make your request take longer any meaningfully measurable time - definetely not comparable to 1ms which extra request would take. IPv4 minimum packet size is 20 bytes (header without data) and add here Modbus request 12 bytes for extra request + that response length, |
Beta Was this translation helpful? Give feedback.
-
|
@alexlansovich was that last example useful? working for you? |
Beta Was this translation helpful? Give feedback.
-
|
I will check this at evening. Thank you. I will also share my full mini project. |
Beta Was this translation helpful? Give feedback.
-
|
I think this is works good. Here is your code https://github.com/alexlansovich/modbus/blob/main/test1.go |
Beta Was this translation helpful? Give feedback.
-
Hello.
I am testing poller function and have problem with splitting registers.
My device can answer only for the 16 registers.
https://datakom.com.tr/upload/Files/500_MODBUS.pdf
The maximum number of registers read in one message is 16. If more registers are requested, the unit will send only the first 16 registers.
I need to read 29. Can you add custom Split?
For example Split(16)
With this way i got error -
Beta Was this translation helpful? Give feedback.
All reactions