bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-07-06 13:42:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #5 from gogf/master

Browse Source 
同步主库
This commit is contained in:
HaiLaz
2019-06-18 09:12:16 +08:00
committed by GitHub
parent 5896dadaad 78ccbeac3b
commit 6c657dff37
36 changed files with 736 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.MD
View File

@ -4,7 +4,6 @@
[![Go Doc](https://godoc.org/github.com/gogf/gf?status.svg)](https://godoc.org/github.com/gogf/gf/g#pkg-subdirectories)
[![Build Status](https://travis-ci.org/gogf/gf.svg?branch=master)](https://travis-ci.org/gogf/gf)
[![Go Report](https://goreportcard.com/badge/github.com/gogf/gf)](https://goreportcard.com/report/github.com/gogf/gf)
[![Code Coverage](https://codecov.io/gh/gogf/gf/branch/master/graph/badge.svg)](https://codecov.io/gh/gogf/gf/branch/master)
[![Production Ready](https://img.shields.io/badge/production-ready-blue.svg)](https://github.com/gogf/gf)
[![License](https://img.shields.io/github/license/gogf/gf.svg?style=flat)](https://github.com/gogf/gf)

1
README_ZH.MD
View File

@ -4,7 +4,6 @@
[![Go Doc](https://godoc.org/github.com/gogf/gf?status.svg)](https://godoc.org/github.com/gogf/gf/g#pkg-subdirectories)
[![Build Status](https://travis-ci.org/gogf/gf.svg?branch=master)](https://travis-ci.org/gogf/gf)
[![Go Report](https://goreportcard.com/badge/github.com/gogf/gf)](https://goreportcard.com/report/github.com/gogf/gf)
[![Code Coverage](https://codecov.io/gh/gogf/gf/branch/master/graph/badge.svg)](https://codecov.io/gh/gogf/gf/branch/master)
[![Production Ready](https://img.shields.io/badge/production-ready-blue.svg)](https://github.com/gogf/gf)
[![License](https://img.shields.io/github/license/gogf/gf.svg?style=flat)](https://github.com/gogf/gf)

1
TODO.MD
View File

@ -50,7 +50,6 @@
1. grpool增加支持阻塞添加任务接口；
# DONE
1. gconv完善针对不同类型的判断，例如：尽量减少sprintf("%v", xxx)来执行string类型的转换；
2. ghttp.Server请求执行中增加服务退出的方法，不再执行后续操作；

12
g/container/glist/glist.go
View File

@ -93,11 +93,9 @@ func (l *List) PopBacks(max int) (values []interface{}) {
if max > 0 && max < length {
length = max
}
tempe := (*Element)(nil)
values = make([]interface{}, length)
for i := 0; i < length; i++ {
tempe = l.list.Back()
values[i] = l.list.Remove(tempe)
values[i] = l.list.Remove(l.list.Back())
}
}
l.mu.Unlock()
@ -107,20 +105,18 @@ func (l *List) PopBacks(max int) (values []interface{}) {
// PopFronts removes <max> elements from front of <l>
// and returns values of the removed elements as slice.
func (l *List) PopFronts(max int) (values []interface{}) {
l.mu.RLock()
l.mu.Lock()
length := l.list.Len()
if length > 0 {
if max > 0 && max < length {
length = max
}
tempe := (*Element)(nil)
values = make([]interface{}, length)
for i := 0; i < length; i++ {
tempe = l.list.Front()
values[i] = l.list.Remove(tempe)
values[i] = l.list.Remove(l.list.Front())
}
}
l.mu.RUnlock()
l.mu.Unlock()
return
}

56
g/container/gqueue/gqueue.go
View File

@ -19,9 +19,9 @@
package gqueue
import (
"github.com/gogf/gf/g/container/glist"
"github.com/gogf/gf/g/container/gtype"
"math"
"github.com/gogf/gf/g/container/glist"
"github.com/gogf/gf/g/container/gtype"
"math"
)
type Queue struct {
@ -35,6 +35,8 @@ type Queue struct {
const (
// Size for queue buffer.
gDEFAULT_QUEUE_SIZE = 10000
// Max batch size per-fetching from list.
gDEFAULT_MAX_BATCH_SIZE = 10
)
// New returns an empty queue object.
@ -66,22 +68,28 @@ func (q *Queue) startAsyncLoop() {
}()
for !q.closed.Val() {
<- q.events
if length := q.list.Len(); length > 0 {
array := make([]interface{}, length)
for i := 0; i < length; i++ {
if e := q.list.Front(); e != nil {
array[i] = q.list.Remove(e)
} else {
break
}
}
for _, v := range array {
// When q.C closes, it will panic here, especially q.C is being blocked for writing.
// It will be caught by recover and be ignored, if any error occurs here.
q.C <- v
}
for !q.closed.Val() {
if length := q.list.Len(); length > 0 {
if length > gDEFAULT_MAX_BATCH_SIZE {
length = gDEFAULT_MAX_BATCH_SIZE
}
for _, v := range q.list.PopFronts(length) {
// When q.C is closed, it will panic here, especially q.C is being blocked for writing.
// If any error occurs here, it will be caught by recover and be ignored.
q.C <- v
}
} else {
break
}
}
// Clear q.events to remain just one event to do the next synchronization check.
for i := 0; i < len(q.events) - 1; i++ {
<- q.events
}
}
// It should be here to close q.C.
// It's the sender's responsibility to close channel when it should be closed.
close(q.C)
}
// Push pushes the data <v> into the queue.
@ -91,7 +99,9 @@ func (q *Queue) Push(v interface{}) {
q.C <- v
} else {
q.list.PushBack(v)
q.events <- struct{}{}
if len(q.events) < gDEFAULT_QUEUE_SIZE {
q.events <- struct{}{}
}
}
}
@ -106,11 +116,11 @@ func (q *Queue) Pop() interface{} {
// which are being blocked reading using Pop method.
func (q *Queue) Close() {
q.closed.Set(true)
if q.events != nil {
close(q.events)
}
if q.C != nil {
close(q.C)
if q.events != nil {
close(q.events)
}
for i := 0; i < gDEFAULT_MAX_BATCH_SIZE; i++ {
q.Pop()
}
}

56
g/container/gqueue/gqueue_unit_test.go Normal file
View File

@ -0,0 +1,56 @@
// Copyright 2017 gf Author(https://github.com/gogf/gf). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
// go test *.go -bench=".*" -benchmem
package gqueue_test
import (
"github.com/gogf/gf/g/container/gqueue"
"github.com/gogf/gf/g/test/gtest"
"testing"
)
func TestQueue_Len(t *testing.T) {
max := 100
for n := 10; n < max; n++ {
q1 := gqueue.New(max)
for i := 0; i < max; i++ {
q1.Push(i)
}
gtest.Assert(q1.Len(), max)
gtest.Assert(q1.Size(), max)
}
}
func TestQueue_Basic(t *testing.T) {
q := gqueue.New()
for i := 0; i < 100; i++ {
q.Push(i)
}
gtest.Assert(q.Pop(), 0)
gtest.Assert(q.Pop(), 1)
}
func TestQueue_Pop(t *testing.T) {
q1 := gqueue.New()
q1.Push(1)
q1.Push(2)
q1.Push(3)
q1.Push(4)
i1 := q1.Pop()
gtest.Assert(i1, 1)
}
func TestQueue_Close(t *testing.T) {
q1 := gqueue.New()
q1.Push(1)
q1.Push(2)
gtest.Assert(q1.Len(), 2)
q1.Close()
}

26
g/container/gtype/bool.go
View File

@ -29,21 +29,33 @@ func NewBool(value...bool) *Bool {
}
// Clone clones and returns a new concurrent-safe object for bool type.
func (t *Bool) Clone() *Bool {
return NewBool(t.Val())
func (v *Bool) Clone() *Bool {
return NewBool(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Bool) Set(value bool) (old bool) {
func (v *Bool) Set(value bool) (old bool) {
if value {
old = atomic.SwapInt32(&t.value, 1) == 1
old = atomic.SwapInt32(&v.value, 1) == 1
} else {
old = atomic.SwapInt32(&t.value, 0) == 1
old = atomic.SwapInt32(&v.value, 0) == 1
}
return
}
// Val atomically loads t.valueue.
func (t *Bool) Val() bool {
return atomic.LoadInt32(&t.value) > 0
func (v *Bool) Val() bool {
return atomic.LoadInt32(&v.value) > 0
}
// Cas executes the compare-and-swap operation for value.
func (v *Bool) Cas(old, new bool) bool {
var oldInt32, newInt32 int32
if old {
oldInt32 = 1
}
if new {
newInt32 = 1
}
return atomic.CompareAndSwapInt32(&v.value, oldInt32, newInt32)
}

21
g/container/gtype/byte.go
View File

@ -26,21 +26,26 @@ func NewByte(value...byte) *Byte {
}
// Clone clones and returns a new concurrent-safe object for byte type.
func (t *Byte) Clone() *Byte {
return NewByte(t.Val())
func (v *Byte) Clone() *Byte {
return NewByte(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Byte) Set(value byte) (old byte) {
return byte(atomic.SwapInt32(&t.value, int32(value)))
func (v *Byte) Set(value byte) (old byte) {
return byte(atomic.SwapInt32(&v.value, int32(value)))
}
// Val atomically loads t.value.
func (t *Byte) Val() byte {
return byte(atomic.LoadInt32(&t.value))
func (v *Byte) Val() byte {
return byte(atomic.LoadInt32(&v.value))
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Byte) Add(delta int) (new byte) {
return byte(atomic.AddInt32(&t.value, int32(delta)))
func (v *Byte) Add(delta byte) (new byte) {
return byte(atomic.AddInt32(&v.value, int32(delta)))
}
// Cas executes the compare-and-swap operation for value.
func (v *Byte) Cas(old, new byte) bool {
return atomic.CompareAndSwapInt32(&v.value, int32(old), int32(new))
}

14
g/container/gtype/bytes.go
View File

@ -23,21 +23,21 @@ func NewBytes(value...[]byte) *Bytes {
}
// Clone clones and returns a new concurrent-safe object for []byte type.
func (t *Bytes) Clone() *Bytes {
return NewBytes(t.Val())
func (v *Bytes) Clone() *Bytes {
return NewBytes(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
// Note: The parameter <value> cannot be nil.
func (t *Bytes) Set(value []byte) (old []byte) {
old = t.Val()
t.value.Store(value)
func (v *Bytes) Set(value []byte) (old []byte) {
old = v.Val()
v.value.Store(value)
return
}
// Val atomically loads t.value.
func (t *Bytes) Val() []byte {
if s := t.value.Load(); s != nil {
func (v *Bytes) Val() []byte {
if s := v.value.Load(); s != nil {
return s.([]byte)
}
return nil

25
g/container/gtype/float32.go
View File

@ -28,27 +28,27 @@ func NewFloat32(value...float32) *Float32 {
}
// Clone clones and returns a new concurrent-safe object for float32 type.
func (t *Float32) Clone() *Float32 {
return NewFloat32(t.Val())
func (v *Float32) Clone() *Float32 {
return NewFloat32(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Float32) Set(value float32) (old float32) {
return math.Float32frombits(atomic.SwapUint32(&t.value, math.Float32bits(value)))
func (v *Float32) Set(value float32) (old float32) {
return math.Float32frombits(atomic.SwapUint32(&v.value, math.Float32bits(value)))
}
// Val atomically loads t.value.
func (t *Float32) Val() float32 {
return math.Float32frombits(atomic.LoadUint32(&t.value))
func (v *Float32) Val() float32 {
return math.Float32frombits(atomic.LoadUint32(&v.value))
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Float32) Add(delta float32) (new float32) {
func (v *Float32) Add(delta float32) (new float32) {
for {
old := math.Float32frombits(t.value)
old := math.Float32frombits(v.value)
new = old + delta
if atomic.CompareAndSwapUint32(
(*uint32)(unsafe.Pointer(&t.value)),
(*uint32)(unsafe.Pointer(&v.value)),
math.Float32bits(old),
math.Float32bits(new),
) {
@ -56,4 +56,9 @@ func (t *Float32) Add(delta float32) (new float32) {
}
}
return
}
}
// Cas executes the compare-and-swap operation for value.
func (v *Float32) Cas(old, new float32) bool {
return atomic.CompareAndSwapUint32(&v.value, uint32(old), uint32(new))
}

23
g/container/gtype/float64.go
View File

@ -28,27 +28,27 @@ func NewFloat64(value...float64) *Float64 {
}
// Clone clones and returns a new concurrent-safe object for float64 type.
func (t *Float64) Clone() *Float64 {
return NewFloat64(t.Val())
func (v *Float64) Clone() *Float64 {
return NewFloat64(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Float64) Set(value float64) (old float64) {
return math.Float64frombits(atomic.SwapUint64(&t.value, math.Float64bits(value)))
func (v *Float64) Set(value float64) (old float64) {
return math.Float64frombits(atomic.SwapUint64(&v.value, math.Float64bits(value)))
}
// Val atomically loads t.value.
func (t *Float64) Val() float64 {
return math.Float64frombits(atomic.LoadUint64(&t.value))
func (v *Float64) Val() float64 {
return math.Float64frombits(atomic.LoadUint64(&v.value))
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Float64) Add(delta float64) (new float64) {
func (v *Float64) Add(delta float64) (new float64) {
for {
old := math.Float64frombits(t.value)
old := math.Float64frombits(v.value)
new = old + delta
if atomic.CompareAndSwapUint64(
(*uint64)(unsafe.Pointer(&t.value)),
(*uint64)(unsafe.Pointer(&v.value)),
math.Float64bits(old),
math.Float64bits(new),
) {
@ -57,3 +57,8 @@ func (t *Float64) Add(delta float64) (new float64) {
}
return
}
// Cas executes the compare-and-swap operation for value.
func (v *Float64) Cas(old, new float64) bool {
return atomic.CompareAndSwapUint64(&v.value, uint64(old), uint64(new))
}

21
g/container/gtype/int.go
View File

@ -26,21 +26,26 @@ func NewInt(value...int) *Int {
}
// Clone clones and returns a new concurrent-safe object for int type.
func (t *Int) Clone() *Int {
return NewInt(t.Val())
func (v *Int) Clone() *Int {
return NewInt(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Int) Set(value int) (old int) {
return int(atomic.SwapInt64(&t.value, int64(value)))
func (v *Int) Set(value int) (old int) {
return int(atomic.SwapInt64(&v.value, int64(value)))
}
// Val atomically loads t.value.
func (t *Int) Val() int {
return int(atomic.LoadInt64(&t.value))
func (v *Int) Val() int {
return int(atomic.LoadInt64(&v.value))
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Int) Add(delta int) (new int) {
return int(atomic.AddInt64(&t.value, int64(delta)))
func (v *Int) Add(delta int) (new int) {
return int(atomic.AddInt64(&v.value, int64(delta)))
}
// Cas executes the compare-and-swap operation for value.
func (v *Int) Cas(old, new int) bool {
return atomic.CompareAndSwapInt64(&v.value, int64(old), int64(new))
}

21
g/container/gtype/int32.go
View File

@ -26,21 +26,26 @@ func NewInt32(value...int32) *Int32 {
}
// Clone clones and returns a new concurrent-safe object for int32 type.
func (t *Int32) Clone() *Int32 {
return NewInt32(t.Val())
func (v *Int32) Clone() *Int32 {
return NewInt32(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Int32) Set(value int32) (old int32) {
return atomic.SwapInt32(&t.value, value)
func (v *Int32) Set(value int32) (old int32) {
return atomic.SwapInt32(&v.value, value)
}
// Val atomically loads t.value.
func (t *Int32) Val() int32 {
return atomic.LoadInt32(&t.value)
func (v *Int32) Val() int32 {
return atomic.LoadInt32(&v.value)
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Int32) Add(delta int32) (new int32) {
return atomic.AddInt32(&t.value, delta)
func (v *Int32) Add(delta int32) (new int32) {
return atomic.AddInt32(&v.value, delta)
}
// Cas executes the compare-and-swap operation for value.
func (v *Int32) Cas(old, new int32) bool {
return atomic.CompareAndSwapInt32(&v.value, old, new)
}

21
g/container/gtype/int64.go
View File

@ -26,21 +26,26 @@ func NewInt64(value...int64) *Int64 {
}
// Clone clones and returns a new concurrent-safe object for int64 type.
func (t *Int64) Clone() *Int64 {
return NewInt64(t.Val())
func (v *Int64) Clone() *Int64 {
return NewInt64(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Int64) Set(value int64) (old int64) {
return atomic.SwapInt64(&t.value, value)
func (v *Int64) Set(value int64) (old int64) {
return atomic.SwapInt64(&v.value, value)
}
// Val atomically loads t.value.
func (t *Int64) Val() int64 {
return atomic.LoadInt64(&t.value)
func (v *Int64) Val() int64 {
return atomic.LoadInt64(&v.value)
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Int64) Add(delta int64) int64 {
return atomic.AddInt64(&t.value, delta)
func (v *Int64) Add(delta int64) int64 {
return atomic.AddInt64(&v.value, delta)
}
// Cas executes the compare-and-swap operation for value.
func (v *Int64) Cas(old, new int64) bool {
return atomic.CompareAndSwapInt64(&v.value, old, new)
}

14
g/container/gtype/interface.go
View File

@ -25,19 +25,19 @@ func NewInterface(value...interface{}) *Interface {
}
// Clone clones and returns a new concurrent-safe object for interface{} type.
func (t *Interface) Clone() *Interface {
return NewInterface(t.Val())
func (v *Interface) Clone() *Interface {
return NewInterface(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
// Note: The parameter <value> cannot be nil.
func (t *Interface) Set(value interface{}) (old interface{}) {
old = t.Val()
t.value.Store(value)
func (v *Interface) Set(value interface{}) (old interface{}) {
old = v.Val()
v.value.Store(value)
return
}
// Val atomically loads t.value.
func (t *Interface) Val() interface{} {
return t.value.Load()
func (v *Interface) Val() interface{} {
return v.value.Load()
}

14
g/container/gtype/string.go
View File

@ -25,20 +25,20 @@ func NewString(value...string) *String {
}
// Clone clones and returns a new concurrent-safe object for string type.
func (t *String) Clone() *String {
return NewString(t.Val())
func (v *String) Clone() *String {
return NewString(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *String) Set(value string) (old string) {
old = t.Val()
t.value.Store(value)
func (v *String) Set(value string) (old string) {
old = v.Val()
v.value.Store(value)
return
}
// Val atomically loads t.value.
func (t *String) Val() string {
s := t.value.Load()
func (v *String) Val() string {
s := v.value.Load()
if s != nil {
return s.(string)
}

21
g/container/gtype/uint.go
View File

@ -26,21 +26,26 @@ func NewUint(value...uint) *Uint {
}
// Clone clones and returns a new concurrent-safe object for uint type.
func (t *Uint) Clone() *Uint {
return NewUint(t.Val())
func (v *Uint) Clone() *Uint {
return NewUint(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Uint) Set(value uint) (old uint) {
return uint(atomic.SwapUint64(&t.value, uint64(value)))
func (v *Uint) Set(value uint) (old uint) {
return uint(atomic.SwapUint64(&v.value, uint64(value)))
}
// Val atomically loads t.value.
func (t *Uint) Val() uint {
return uint(atomic.LoadUint64(&t.value))
func (v *Uint) Val() uint {
return uint(atomic.LoadUint64(&v.value))
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Uint) Add(delta uint) (new uint) {
return uint(atomic.AddUint64(&t.value, uint64(delta)))
func (v *Uint) Add(delta uint) (new uint) {
return uint(atomic.AddUint64(&v.value, uint64(delta)))
}
// Cas executes the compare-and-swap operation for value.
func (v *Uint) Cas(old, new uint) bool {
return atomic.CompareAndSwapUint64(&v.value, uint64(old), uint64(new))
}

21
g/container/gtype/uint32.go
View File

@ -26,21 +26,26 @@ func NewUint32(value...uint32) *Uint32 {
}
// Clone clones and returns a new concurrent-safe object for uint32 type.
func (t *Uint32) Clone() *Uint32 {
return NewUint32(t.Val())
func (v *Uint32) Clone() *Uint32 {
return NewUint32(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Uint32) Set(value uint32) (old uint32) {
return atomic.SwapUint32(&t.value, value)
func (v *Uint32) Set(value uint32) (old uint32) {
return atomic.SwapUint32(&v.value, value)
}
// Val atomically loads t.value.
func (t *Uint32) Val() uint32 {
return atomic.LoadUint32(&t.value)
func (v *Uint32) Val() uint32 {
return atomic.LoadUint32(&v.value)
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Uint32) Add(delta uint32) (new uint32) {
return atomic.AddUint32(&t.value, delta)
func (v *Uint32) Add(delta uint32) (new uint32) {
return atomic.AddUint32(&v.value, delta)
}
// Cas executes the compare-and-swap operation for value.
func (v *Uint32) Cas(old, new uint32) bool {
return atomic.CompareAndSwapUint32(&v.value, old, new)
}

21
g/container/gtype/uint64.go
View File

@ -26,21 +26,26 @@ func NewUint64(value...uint64) *Uint64 {
}
// Clone clones and returns a new concurrent-safe object for uint64 type.
func (t *Uint64) Clone() *Uint64 {
return NewUint64(t.Val())
func (v *Uint64) Clone() *Uint64 {
return NewUint64(v.Val())
}
// Set atomically stores <value> into t.value and returns the previous value of t.value.
func (t *Uint64) Set(value uint64) (old uint64) {
return atomic.SwapUint64(&t.value, value)
func (v *Uint64) Set(value uint64) (old uint64) {
return atomic.SwapUint64(&v.value, value)
}
// Val atomically loads t.value.
func (t *Uint64) Val() uint64 {
return atomic.LoadUint64(&t.value)
func (v *Uint64) Val() uint64 {
return atomic.LoadUint64(&v.value)
}
// Add atomically adds <delta> to t.value and returns the new value.
func (t *Uint64) Add(delta uint64) (new uint64) {
return atomic.AddUint64(&t.value, delta)
func (v *Uint64) Add(delta uint64) (new uint64) {
return atomic.AddUint64(&v.value, delta)
}
// Cas executes the compare-and-swap operation for value.
func (v *Uint64) Cas(old, new uint64) bool {
return atomic.CompareAndSwapUint64(&v.value, old, new)
}

189
g/container/gvar/gvar.go
View File

@ -8,104 +8,179 @@
package gvar
import (
"github.com/gogf/gf/g/container/gtype"
"github.com/gogf/gf/g/os/gtime"
"github.com/gogf/gf/g/util/gconv"
"time"
"github.com/gogf/gf/g/container/gtype"
"github.com/gogf/gf/g/os/gtime"
"github.com/gogf/gf/g/util/gconv"
"time"
)
// Var is an universal variable type.
type Var struct {
value interface{} // Underlying value.
safe bool // Concurrent safe or not.
value interface{} // Underlying value.
safe bool // Concurrent safe or not.
}
// New returns a new Var with given <value>.
// The parameter <unsafe> used to specify whether using Var in un-concurrent-safety,
// which is false in default, means concurrent-safe.
func New(value interface{}, unsafe...bool) *Var {
v := &Var{}
if len(unsafe) == 0 || !unsafe[0] {
v.safe = true
v.value = gtype.NewInterface(value)
} else {
v.value = value
}
return v
func New(value interface{}, unsafe ...bool) *Var {
v := &Var{}
if len(unsafe) == 0 || !unsafe[0] {
v.safe = true
v.value = gtype.NewInterface(value)
} else {
v.value = value
}
return v
}
// Set sets <value> to <v>, and returns the old value.
func (v *Var) Set(value interface{}) (old interface{}) {
if v.safe {
old = v.value.(*gtype.Interface).Set(value)
} else {
old = v.value
v.value = value
}
return
if v.safe {
old = v.value.(*gtype.Interface).Set(value)
} else {
old = v.value
v.value = value
}
return
}
// Val returns the current value of <v>.
func (v *Var) Val() interface{} {
if v.safe {
return v.value.(*gtype.Interface).Val()
} else {
return v.value
}
if v.safe {
return v.value.(*gtype.Interface).Val()
}
return v.value
}
// See Val().
// Interface is alias of Val.
func (v *Var) Interface() interface{} {
return v.Val()
return v.Val()
}
// Time converts and returns <v> as time.Time.
// The parameter <format> specifies the format of the time string using gtime,
// eg: Y-m-d H:i:s.
func (v *Var) Time(format...string) time.Time {
return gconv.Time(v.Val(), format...)
func (v *Var) Time(format ...string) time.Time {
return gconv.Time(v.Val(), format...)
}
// TimeDuration converts and returns <v> as time.Duration.
// Duration converts and returns <v> as time.Duration.
// If value of <v> is string, then it uses time.ParseDuration for conversion.
func (v *Var) Duration() time.Duration {
return gconv.Duration(v.Val())
return gconv.Duration(v.Val())
}
// GTime converts and returns <v> as *gtime.Time.
// The parameter <format> specifies the format of the time string using gtime,
// eg: Y-m-d H:i:s.
func (v *Var) GTime(format...string) *gtime.Time {
return gconv.GTime(v.Val(), format...)
func (v *Var) GTime(format ...string) *gtime.Time {
return gconv.GTime(v.Val(), format...)
}
// Struct maps value of <v> to <objPointer>.
// The parameter <objPointer> should be a pointer to a struct instance.
// The parameter <mapping> is used to specify the key-to-attribute mapping rules.
func (v *Var) Struct(pointer interface{}, mapping...map[string]string) error {
return gconv.Struct(v.Val(), pointer, mapping...)
func (v *Var) Struct(pointer interface{}, mapping ...map[string]string) error {
return gconv.Struct(v.Val(), pointer, mapping...)
}
func (v *Var) IsNil() bool { return v.Val() == nil }
func (v *Var) Bytes() []byte { return gconv.Bytes(v.Val()) }
func (v *Var) String() string { return gconv.String(v.Val()) }
func (v *Var) Bool() bool { return gconv.Bool(v.Val()) }
// IsNil checks whether <v> is nil.
func (v *Var) IsNil() bool {
return v.Val() == nil
}
func (v *Var) Int() int { return gconv.Int(v.Val()) }
func (v *Var) Int8() int8 { return gconv.Int8(v.Val()) }
func (v *Var) Int16() int16 { return gconv.Int16(v.Val()) }
func (v *Var) Int32() int32 { return gconv.Int32(v.Val()) }
func (v *Var) Int64() int64 { return gconv.Int64(v.Val()) }
// Bytes converts and returns <v> as []byte.
func (v *Var) Bytes() []byte {
return gconv.Bytes(v.Val())
}
func (v *Var) Uint() uint { return gconv.Uint(v.Val()) }
func (v *Var) Uint8() uint8 { return gconv.Uint8(v.Val()) }
func (v *Var) Uint16() uint16 { return gconv.Uint16(v.Val()) }
func (v *Var) Uint32() uint32 { return gconv.Uint32(v.Val()) }
func (v *Var) Uint64() uint64 { return gconv.Uint64(v.Val()) }
// String converts and returns <v> as string.
func (v *Var) String() string {
return gconv.String(v.Val())
}
func (v *Var) Float32() float32 { return gconv.Float32(v.Val()) }
func (v *Var) Float64() float64 { return gconv.Float64(v.Val()) }
// Bool converts and returns <v> as bool.
func (v *Var) Bool() bool {
return gconv.Bool(v.Val())
}
func (v *Var) Ints() []int { return gconv.Ints(v.Val()) }
func (v *Var) Floats() []float64 { return gconv.Floats(v.Val()) }
func (v *Var) Strings() []string { return gconv.Strings(v.Val()) }
func (v *Var) Interfaces() []interface{} { return gconv.Interfaces(v.Val()) }
// Int converts and returns <v> as int.
func (v *Var) Int() int {
return gconv.Int(v.Val())
}
// Int8 converts and returns <v> as int8.
func (v *Var) Int8() int8 {
return gconv.Int8(v.Val())
}
// Int16 converts and returns <v> as int16.
func (v *Var) Int16() int16 {
return gconv.Int16(v.Val())
}
// Int32 converts and returns <v> as int32.
func (v *Var) Int32() int32 {
return gconv.Int32(v.Val())
}
// Int64 converts and returns <v> as int64.
func (v *Var) Int64() int64 {
return gconv.Int64(v.Val())
}
// Uint converts and returns <v> as uint.
func (v *Var) Uint() uint {
return gconv.Uint(v.Val())
}
// Uint8 converts and returns <v> as uint8.
func (v *Var) Uint8() uint8 {
return gconv.Uint8(v.Val())
}
// Uint16 converts and returns <v> as uint16.
func (v *Var) Uint16() uint16 {
return gconv.Uint16(v.Val())
}
// Uint32 converts and returns <v> as uint32.
func (v *Var) Uint32() uint32 {
return gconv.Uint32(v.Val())
}
// Uint64 converts and returns <v> as uint64.
func (v *Var) Uint64() uint64 {
return gconv.Uint64(v.Val())
}
// Float32 converts and returns <v> as float32.
func (v *Var) Float32() float32 {
return gconv.Float32(v.Val())
}
// Float64 converts and returns <v> as float64.
func (v *Var) Float64() float64 {
return gconv.Float64(v.Val())
}
// Ints converts and returns <v> as []int.
func (v *Var) Ints() []int {
return gconv.Ints(v.Val())
}
// Floats converts and returns <v> as []float64.
func (v *Var) Floats() []float64 {
return gconv.Floats(v.Val())
}
// Strings converts and returns <v> as []string.
func (v *Var) Strings() []string {
return gconv.Strings(v.Val())
}
// Interfaces converts and returns <v> as []interfaces{}.
func (v *Var) Interfaces() []interface{} {
return gconv.Interfaces(v.Val())
}

149
g/os/gfpool/gfpool_z_unit_test.go Normal file
View File

@ -0,0 +1,149 @@
package gfpool_test
import (
"github.com/gogf/gf/g/os/gfile"
"github.com/gogf/gf/g/os/gfpool"
"github.com/gogf/gf/g/os/glog"
"github.com/gogf/gf/g/test/gtest"
"os"
"testing"
"time"
)
// TestOpen test open file cache
func TestOpen(t *testing.T) {
testFile := start("TestOpen.txt")
gtest.Case(t, func() {
f, err := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666)
gtest.AssertEQ(err, nil)
f.Close()
f2, err1 := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666)
gtest.AssertEQ(err1, nil)
gtest.AssertEQ(f, f2)
f2.Close()
// Deprecated test
f3, err2 := gfpool.OpenFile(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666)
gtest.AssertEQ(err2, nil)
gtest.AssertEQ(f, f3)
f3.Close()
})
stop(testFile)
}
// TestOpenErr test open file error
func TestOpenErr(t *testing.T) {
gtest.Case(t, func() {
testErrFile := "errorPath"
_, err := gfpool.Open(testErrFile, os.O_RDWR, 0666)
gtest.AssertNE(err, nil)
// delete file error
testFile := start("TestOpenDeleteErr.txt")
pool := gfpool.New(testFile, os.O_RDWR, 0666)
_, err1 := pool.File()
gtest.AssertEQ(err1, nil)
stop(testFile)
_, err1 = pool.File()
gtest.AssertNE(err1, nil)
// append mode delete file error and create again
testFile = start("TestOpenCreateErr.txt")
pool = gfpool.New(testFile, os.O_CREATE, 0666)
_, err1 = pool.File()
gtest.AssertEQ(err1, nil)
stop(testFile)
_, err1 = pool.File()
gtest.AssertEQ(err1, nil)
// append mode delete file error
testFile = start("TestOpenAppendErr.txt")
pool = gfpool.New(testFile, os.O_APPEND, 0666)
_, err1 = pool.File()
gtest.AssertEQ(err1, nil)
stop(testFile)
_, err1 = pool.File()
gtest.AssertNE(err1, nil)
// trunc mode delete file error
testFile = start("TestOpenTruncErr.txt")
pool = gfpool.New(testFile, os.O_TRUNC, 0666)
_, err1 = pool.File()
gtest.AssertEQ(err1, nil)
stop(testFile)
_, err1 = pool.File()
gtest.AssertNE(err1, nil)
})
}
// TestOpenExpire test open file cache expire
func TestOpenExpire(t *testing.T) {
testFile := start("TestOpenExpire.txt")
gtest.Case(t, func() {
f, err := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666, 100)
gtest.AssertEQ(err, nil)
f.Close()
time.Sleep(150 * time.Millisecond)
f2, err1 := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666, 100)
gtest.AssertEQ(err1, nil)
gtest.AssertNE(f, f2)
f2.Close()
// Deprecated test
f3, err2 := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666, 100)
gtest.AssertEQ(err2, nil)
gtest.AssertEQ(f2, f3)
f3.Close()
})
stop(testFile)
}
// TestNewPool test gfpool new function
func TestNewPool(t *testing.T) {
testFile := start("TestNewPool.txt")
gtest.Case(t, func() {
f, err := gfpool.Open(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666)
gtest.AssertEQ(err, nil)
f.Close()
pool := gfpool.New(testFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC|os.O_APPEND, 0666)
f2, err1 := pool.File()
// pool not equal
gtest.AssertEQ(err1, nil)
gtest.AssertNE(f, f2)
f2.Close()
pool.Close()
})
stop(testFile)
}
// test before
func start(name string) string {
testFile := os.TempDir() + string(os.PathSeparator) + name
if gfile.Exists(testFile) {
gfile.Remove(testFile)
}
content := "123"
gfile.PutContents(testFile, content)
return testFile
}
// test after
func stop(testFile string) {
if gfile.Exists(testFile) {
err := gfile.Remove(testFile)
if err != nil {
glog.Error(err)
}
}
}

6
g/os/glog/glog.go
View File

@ -37,8 +37,8 @@ func init() {
}
// SetPath sets the directory path for file logging.
func SetPath(path string) {
logger.SetPath(path)
func SetPath(path string) error {
return logger.SetPath(path)
}
// GetPath returns the logging directory path for file logging.
@ -89,7 +89,7 @@ func SetAsync(enabled bool) {
logger.SetAsync(enabled)
}
// SetStdoutPrint sets whether ouptput the logging contents to stdout, which is false in default.
// SetStdoutPrint sets whether ouptput the logging contents to stdout, which is true in default.
func SetStdoutPrint(enabled bool) {
logger.SetStdoutPrint(enabled)
}

61
g/os/gmlock/gmlock_mutex.go
View File

@ -15,17 +15,18 @@ import (
// It wraps the sync.RWMutex to implements more rich features.
type Mutex struct {
mu sync.RWMutex
wid *gtype.Int64 // Unique id, used for multiple safely Unlock.
rcount *gtype.Int // Reading locks count.
wcount *gtype.Int // Writing locks count.
wid *gtype.Int64 // Unique id, used for multiple safely Unlock.
number *gtype.Int // Locking number.
// 0: writing lock false;
// 1: writing lock true;
// >=2: reading lock;
}
// NewMutex creates and returns a new mutex.
func NewMutex() *Mutex {
return &Mutex{
wid : gtype.NewInt64(),
rcount : gtype.NewInt(),
wcount : gtype.NewInt(),
number : gtype.NewInt(),
}
}
@ -33,29 +34,25 @@ func NewMutex() *Mutex {
// If the lock is already locked for reading or writing,
// Lock blocks until the lock is available.
func (m *Mutex) Lock() {
m.wcount.Add(1)
m.mu.Lock()
m.wid.Add(1)
m.mu.Lock()
m.number.Set(1)
m.wid.Add(1)
}
// Unlock unlocks the write lock.
// It is safe to be called multiple times.
func (m *Mutex) Unlock() {
if m.wcount.Val() > 0 {
if m.wcount.Add(-1) >= 0 {
m.mu.Unlock()
} else {
m.wcount.Add(1)
}
}
if m.number.Cas(1, 0) {
m.mu.Unlock()
}
}
// RLock locks mutex for reading.
// If the mutex is already locked for writing,
// It blocks until the lock is available.
func (m *Mutex) RLock() {
m.rcount.Add(1)
m.mu.RLock()
m.number.Add(2)
}
// RUnlock undoes a single RLock call;
@ -64,11 +61,11 @@ func (m *Mutex) RLock() {
// on entry to RUnlock.
// It is safe to be called multiple times.
func (m *Mutex) RUnlock() {
if m.rcount.Val() > 0 {
if m.rcount.Add(-1) >= 0 {
if n := m.number.Val(); n >= 2 && n%2 == 0 {
if n := m.number.Add(-2); n >= 0 && n%2 == 0 {
m.mu.RUnlock()
} else {
m.rcount.Add(1)
m.number.Add(2)
}
}
}
@ -77,27 +74,25 @@ func (m *Mutex) RUnlock() {
// It returns true if success, or if there's a write/read lock on the mutex,
// it returns false.
func (m *Mutex) TryLock() bool {
// The first check, but it cannot ensure the atomicity.
if m.wcount.Val() == 0 && m.rcount.Val() == 0 {
// The second check, it ensures the atomicity with atomic Add.
if m.wcount.Add(1) == 1 {
m.mu.Lock()
m.wid.Add(1)
return true
} else {
m.wcount.Add(-1)
}
if m.number.Cas(0, 1) {
m.mu.Lock()
m.wid.Add(1)
return true
}
return false
}
// TryRLock tries locking the mutex for reading.
// It returns true if success, or if there's a write lock on the mutex, it returns false.
// It returns true if success, or if there's a writing lock on the mutex, it returns false.
//
// Note that it's not an atomic operation.
//
// TODO It should be an atomic operation.
func (m *Mutex) TryRLock() bool {
// There must be no write lock on mutex.
if m.wcount.Val() == 0 {
m.rcount.Add(1)
// There must be no writing lock on the mutex.
if n := m.number.Val(); n%2 == 0 {
m.mu.RLock()
m.number.Add(2)
return true
}
return false

12
g/os/gproc/gproc.go
View File

@ -4,11 +4,7 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
// Package gproc implements communication and management of processes.
//
// 进程管理/通信,
// 本进程管理从syscall, os.StartProcess, exec.Cmd都使用过,
// 最后采用了exec.Cmd来实现多进程管理，这是一个顶层的跨平台封装，兼容性更好，另外两个是偏底层的接口。
// Package gproc implements management and communication for processes.
package gproc
import (
@ -62,11 +58,11 @@ func IsChild() bool {
}
// 设置gproc父进程ID，当ppid为0时表示该进程为gproc主进程，否则为gproc子进程
func SetPPid(ppid int) {
func SetPPid(ppid int) error {
if ppid > 0 {
os.Setenv(gPROC_ENV_KEY_PPID_KEY, gconv.String(ppid))
return os.Setenv(gPROC_ENV_KEY_PPID_KEY, gconv.String(ppid))
} else {
os.Unsetenv(gPROC_ENV_KEY_PPID_KEY)
return os.Unsetenv(gPROC_ENV_KEY_PPID_KEY)
}
}

15
g/os/gproc/gproc_comm.go
View File

@ -15,18 +15,21 @@ import (
"github.com/gogf/gf/g/container/gmap"
)
// 进程通信数据结构
type Msg struct {
SendPid int `json:"spid"` // 发送进程ID
RecvPid int `json:"rpid"` // 接收进程ID
Group string `json:"group"` // 分组名称
Data []byte `json:"data"` // 原始数据
}
// 本地进程通信接收消息队列(按照分组进行构建的map，键值为*gqueue.Queue对象)
var commReceiveQueues = gmap.NewStrAnyMap()
// (用于发送)已建立的PID对应的Conn通信对象，键值为一个Pool，防止并行使用同一个通信对象造成数据重叠
var commPidConnMap = gmap.NewIntAnyMap()
// TCP通信数据结构定义
type Msg struct {
Pid int // PID，来源哪个进程
Data []byte // 数据
Group string // 分组名称
}
// 获取指定进程的通信文件地址
func getCommFilePath(pid int) string {

103
g/os/gproc/gproc_comm_receive.go
View File

@ -10,15 +10,15 @@
package gproc
import (
"fmt"
"net"
"github.com/gogf/gf/g/os/glog"
"github.com/gogf/gf/g/net/gtcp"
"github.com/gogf/gf/g/os/gfile"
"github.com/gogf/gf/g/util/gconv"
"github.com/gogf/gf/g/encoding/gbinary"
"github.com/gogf/gf/g/container/gqueue"
"github.com/gogf/gf/g/container/gtype"
"encoding/json"
"fmt"
"github.com/gogf/gf/g/container/gqueue"
"github.com/gogf/gf/g/container/gtype"
"github.com/gogf/gf/g/net/gtcp"
"github.com/gogf/gf/g/os/gfile"
"github.com/gogf/gf/g/os/glog"
"github.com/gogf/gf/g/util/gconv"
"net"
)
const (
@ -35,7 +35,7 @@ var (
// 进程只有在执行该方法后才会打开请求端口，默认情况下不允许进程间通信。
func Receive(group...string) *Msg {
// 一个进程只能开启一个监听goroutine
if tcpListened.Set(true) == false {
if !tcpListened.Val() && tcpListened.Set(true) == false {
go startTcpListening()
}
queue := (*gqueue.Queue)(nil)
@ -75,7 +75,9 @@ func startTcpListening() {
continue
}
// 将监听的端口保存到通信文件中(字符串类型存放)
gfile.PutContents(getCommFilePath(Pid()), gconv.String(i))
if err := gfile.PutContents(getCommFilePath(Pid()), gconv.String(i)); err != nil {
glog.Error(err)
}
break
}
for {
@ -89,74 +91,41 @@ func startTcpListening() {
// TCP数据通信处理回调函数
func tcpServiceHandler(conn *gtcp.Conn) {
retry := gtcp.Retry {
Count : 3,
Interval: 10,
}
option := gtcp.PkgOption{
Retry : gtcp.Retry {
Count : 3,
Interval: 10,
},
}
for {
var result []byte
buffer, err := conn.Recv(-1, retry)
buffer, err := conn.RecvPkg(option)
if len(buffer) > 0 {
var msgs []*Msg
for _, msg := range bufferToMsgs(buffer) {
if v := commReceiveQueues.Get(msg.Group); v != nil {
msgs = append(msgs, msg)
} else {
result = []byte(fmt.Sprintf("group [%s] does not exist", msg.Group))
break
}
msg := new(Msg)
if err := json.Unmarshal(buffer, msg); err != nil {
glog.Error(err)
continue
}
// 成功时会返回ok给peer
if len(result) == 0 {
if v := commReceiveQueues.Get(msg.Group); v == nil {
result = []byte(fmt.Sprintf("group [%s] does not exist", msg.Group))
break
} else {
result = []byte("ok")
for _, msg := range msgs {
if v := commReceiveQueues.Get(msg.Group); v != nil {
v.(*gqueue.Queue).Push(msg)
}
if v := commReceiveQueues.Get(msg.Group); v != nil {
v.(*gqueue.Queue).Push(msg)
}
}
}
// 产生错误(或者对方已经关闭链接)时，退出接收循环
if err == nil {
conn.Send(result, retry)
if err := conn.SendPkg(result, option); err != nil {
glog.Error(err)
}
} else {
conn.Close()
if err := conn.Close(); err != nil {
glog.Error(err)
}
return
}
}
}
// 数据解包，防止黏包
// 数据格式：总长度(24bit)|发送进程PID(24bit)|接收进程PID(24bit)|分组长度(8bit)|分组名称(变长)|校验(32bit)|参数(变长)
func bufferToMsgs(buffer []byte) []*Msg {
s := 0
msgs := make([]*Msg, 0)
for s < len(buffer) {
// 长度解析及校验
length := gbinary.DecodeToInt(buffer[s : s + 3])
if length < 14 || length > len(buffer) {
s++
continue
}
// 分组信息解析
groupLen := gbinary.DecodeToInt(buffer[s + 9 : s + 10])
// checksum校验(仅对参数做校验，提高校验效率)
checksum1 := gbinary.DecodeToUint32(buffer[s + 10 + groupLen : s + 10 + groupLen + 4])
checksum2 := gtcp.Checksum(buffer[s + 10 + groupLen + 4 : s + length])
if checksum1 != checksum2 {
s++
continue
}
// 接收进程PID校验
if Pid() == gbinary.DecodeToInt(buffer[s + 6 : s + 9]) {
msgs = append(msgs, &Msg {
Pid : gbinary.DecodeToInt(buffer[s + 3 : s + 6]),
Data : buffer[s + 10 + groupLen + 4 : s + length],
Group : string(buffer[s + 10 : s + 10 + groupLen]),
})
}
s += length
}
return msgs
}

91
g/os/gproc/gproc_comm_send.go
View File

@ -7,16 +7,15 @@
package gproc
import (
"bytes"
"errors"
"fmt"
"github.com/gogf/gf/g/encoding/gbinary"
"github.com/gogf/gf/g/net/gtcp"
"github.com/gogf/gf/g/os/gfcache"
"github.com/gogf/gf/g/os/glog"
"github.com/gogf/gf/g/util/gconv"
"io"
"time"
"bytes"
"encoding/json"
"errors"
"fmt"
"github.com/gogf/gf/g/net/gtcp"
"github.com/gogf/gf/g/os/gfcache"
"github.com/gogf/gf/g/util/gconv"
"io"
"time"
)
const (
@ -27,53 +26,53 @@ const (
)
// 向指定gproc进程发送数据.
// 数据格式：总长度(24bit)|发送进程PID(24bit)|接收进程PID(24bit)|分组长度(8bit)|分组名称(变长)|校验(32bit)|参数(变长)
func Send(pid int, data []byte, group...string) error {
groupName := gPROC_COMM_DEAFULT_GRUOP_NAME
if len(group) > 0 {
groupName = group[0]
}
buffer := make([]byte, 0)
buffer = append(buffer, gbinary.EncodeByLength(3, len(groupName) + len(data) + 14)...)
buffer = append(buffer, gbinary.EncodeByLength(3, Pid())...)
buffer = append(buffer, gbinary.EncodeByLength(3, pid)...)
buffer = append(buffer, gbinary.EncodeByLength(1, len(groupName))...)
buffer = append(buffer, []byte(groupName)...)
buffer = append(buffer, gbinary.EncodeUint32(gtcp.Checksum(data))...)
buffer = append(buffer, data...)
// 执行发送流程
var err error
msg := Msg{
SendPid : Pid(),
RecvPid : pid,
Group : gPROC_COMM_DEAFULT_GRUOP_NAME,
Data : data,
}
if len(group) > 0 {
msg.Group = group[0]
}
msgBytes, err := json.Marshal(msg)
if err != nil {
return err
}
var buf []byte
var conn *gtcp.Conn
for i := gPROC_COMM_FAILURE_RETRY_COUNT; i > 0; i-- {
if conn, err = getConnByPid(pid); err == nil {
defer conn.Close()
buf, err = conn.SendRecvWithTimeout(buffer, -1, gPROC_COMM_SEND_TIMEOUT*time.Millisecond)
if len(buf) > 0 {
// 如果有返回值，如果不是"ok"，那么表示是错误信息
if !bytes.EqualFold(buf, []byte("ok")) {
err = errors.New(string(buf))
break
}
}
// EOF不算异常错误
if err == nil || err == io.EOF {
break
} else {
glog.Error(err)
}
var conn *gtcp.PoolConn
// 循环获取连接TCP对象
for i := gPROC_COMM_FAILURE_RETRY_COUNT; i > 0; i-- {
if conn, err = getConnByPid(pid); err == nil {
break
}
time.Sleep(gPROC_COMM_FAILURE_RETRY_TIMEOUT*time.Millisecond)
}
if conn == nil {
return err
}
defer conn.Close()
// 执行数据发送
buf, err = conn.SendRecvPkgWithTimeout(msgBytes, gPROC_COMM_SEND_TIMEOUT*time.Millisecond)
if len(buf) > 0 {
if !bytes.EqualFold(buf, []byte("ok")) {
err = errors.New(string(buf))
}
time.Sleep(gPROC_COMM_FAILURE_RETRY_TIMEOUT*time.Millisecond)
}
// EOF不算异常错误
if err == io.EOF {
err = nil
}
return err
}
// 获取指定进程的TCP通信对象
func getConnByPid(pid int) (*gtcp.Conn, error) {
func getConnByPid(pid int) (*gtcp.PoolConn, error) {
port := getPortByPid(pid)
if port > 0 {
if conn, err := gtcp.NewConn(fmt.Sprintf("127.0.0.1:%d", port)); err == nil {
if conn, err := gtcp.NewPoolConn(fmt.Sprintf("127.0.0.1:%d", port)); err == nil {
return conn, nil
} else {
return nil, err

6
g/os/gview/gview.go
View File

@ -20,6 +20,7 @@ import (
"sync"
)
// View object for template engine.
type View struct {
mu sync.RWMutex
paths *garray.StringArray // Searching path array.
@ -28,10 +29,10 @@ type View struct {
delimiters []string // Customized template delimiters.
}
// Template params type.
// Params is type for template params.
type Params = map[string]interface{}
// Customized template function map type.
// FuncMap is type for custom template functions.
type FuncMap = map[string]interface{}
// Default view object.
@ -218,7 +219,6 @@ func (view *View) AddPath(path string) error {
return nil
}
view.paths.Append(realPath)
//glog.Debug("[gview] AddPath:", realPath)
return nil
}

18
g/test/gtest/gtest_test.go Normal file
View File

@ -0,0 +1,18 @@
// Copyright 2018 gf Author(https://github.com/gogf/gf). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package gtest_test
import (
"github.com/gogf/gf/g/test/gtest"
"testing"
)
func TestCase(t *testing.T) {
gtest.Case(t, func() {
gtest.Assert(1, 1)
})
}

2
geg/net/gtcp/pkg_operations/common/gtcp_common_server.go
View File

@ -30,7 +30,7 @@ func main() {
case "heartbeat":
onClientHeartBeat(conn, msg)
default:
glog.Errorfln("invalid message: %v", msg)
glog.Errorf("invalid message: %v", msg)
break
}
}

4
geg/os/gflock/gflock.go
View File

@ -10,8 +10,8 @@ import (
func main() {
l := gflock.New("demo.lock")
l.Lock()
glog.Printfln("locked by pid: %d", gproc.Pid())
glog.Printf("locked by pid: %d", gproc.Pid())
time.Sleep(3 * time.Second)
l.UnLock()
glog.Printfln("unlocked by pid: %d", gproc.Pid())
glog.Printf("unlocked by pid: %d", gproc.Pid())
}

10
geg/os/gproc/gproc.go
View File

@ -10,18 +10,18 @@ import (
// 父子进程基本演示
func main() {
if gproc.IsChild() {
glog.Printfln("%d: Hi, I am child, waiting 3 seconds to die", gproc.Pid())
glog.Printf("%d: Hi, I am child, waiting 3 seconds to die", gproc.Pid())
time.Sleep(time.Second)
glog.Printfln("%d: 1", gproc.Pid())
glog.Printf("%d: 1", gproc.Pid())
time.Sleep(time.Second)
glog.Printfln("%d: 2", gproc.Pid())
glog.Printf("%d: 2", gproc.Pid())
time.Sleep(time.Second)
glog.Printfln("%d: 3", gproc.Pid())
glog.Printf("%d: 3", gproc.Pid())
} else {
m := gproc.NewManager()
p := m.NewProcess(os.Args[0], os.Args, os.Environ())
p.Start()
p.Wait()
glog.Printfln("%d: child died", gproc.Pid())
glog.Printf("%d: child died", gproc.Pid())
}
}

16
geg/os/gproc/gproc3.go
View File

@ -11,25 +11,25 @@ import (
// 请使用go build编译后运行，不要使用IDE运行，因为IDE大多采用的是子进程方式执行。
func main() {
if gproc.IsChild() {
glog.Printfln("%d: I am child, waiting 10 seconds to die", gproc.Pid())
glog.Printf("%d: I am child, waiting 10 seconds to die", gproc.Pid())
//p, err := os.FindProcess(os.Getppid())
//fmt.Println(err)
//p.Kill()
time.Sleep(2 * time.Second)
glog.Printfln("%d: 2", gproc.Pid())
glog.Printf("%d: 2", gproc.Pid())
time.Sleep(2 * time.Second)
glog.Printfln("%d: 4", gproc.Pid())
glog.Printf("%d: 4", gproc.Pid())
time.Sleep(2 * time.Second)
glog.Printfln("%d: 6", gproc.Pid())
glog.Printf("%d: 6", gproc.Pid())
time.Sleep(2 * time.Second)
glog.Printfln("%d: 8", gproc.Pid())
glog.Printf("%d: 8", gproc.Pid())
time.Sleep(2 * time.Second)
glog.Printfln("%d: died", gproc.Pid())
glog.Printf("%d: died", gproc.Pid())
} else {
p := gproc.NewProcess(os.Args[0], os.Args, os.Environ())
p.Start()
glog.Printfln("%d: I am main, waiting 3 seconds to die", gproc.Pid())
glog.Printf("%d: I am main, waiting 3 seconds to die", gproc.Pid())
time.Sleep(3 * time.Second)
glog.Printfln("%d: died", gproc.Pid())
glog.Printf("%d: died", gproc.Pid())
}
}

2
geg/os/gproc/gproc4.go
View File

@ -11,7 +11,7 @@ import (
// 查看父子进程的环境变量
func main() {
time.Sleep(5 * time.Second)
glog.Printfln("%d: %v", gproc.Pid(), genv.All())
glog.Printf("%d: %v", gproc.Pid(), genv.All())
p := gproc.NewProcess(os.Args[0], os.Args, os.Environ())
p.Start()
}

6
geg/os/gproc/gproc_comm.go
View File

@ -8,6 +8,7 @@ import (
"github.com/gogf/gf/g/os/glog"
"github.com/gogf/gf/g/os/gproc"
"github.com/gogf/gf/g/os/gtime"
"github.com/gogf/gf/g/os/gtimer"
"os"
"time"
)
@ -15,11 +16,10 @@ import (
func main() {
fmt.Printf("%d: I am child? %v\n", gproc.Pid(), gproc.IsChild())
if gproc.IsChild() {
gtime.SetInterval(time.Second, func() bool {
gtimer.SetInterval(time.Second, func() {
if err := gproc.Send(gproc.PPid(), []byte(gtime.Datetime())); err != nil {
glog.Error(err)
}
return true
})
select {}
} else {
@ -28,7 +28,7 @@ func main() {
p.Start()
for {
msg := gproc.Receive()
fmt.Printf("%d: receive from %d, data: %s\n", gproc.Pid(), msg.Pid, string(msg.Data))
fmt.Printf("%d: receive from %d, data: %s\n", gproc.Pid(), msg.SendPid, string(msg.Data))
}
}
}

44
geg/other/test.go
View File

@ -1,17 +1,45 @@
package main
import (
"github.com/gogf/gf/g/container/gqueue"
"github.com/gogf/gf/g/container/garray"
"github.com/gogf/gf/g/os/gmlock"
"github.com/gogf/gf/g/test/gtest"
"time"
)
func main() {
max := 100
q := gqueue.New(max)
for i := 1; i < max; i++ {
q.Push(i)
}
q.Close()
gtest.Assert(q.Len(), 1)
mu := gmlock.NewMutex()
array := garray.New()
go func() {
mu.LockFunc(func() {
array.Append(1)
time.Sleep(100 * time.Millisecond)
})
}()
go func() {
time.Sleep(50 * time.Millisecond)
mu.LockFunc(func() {
array.Append(1)
})
}()
go func() {
time.Sleep(50 * time.Millisecond)
mu.LockFunc(func() {
array.Append(1)
})
}()
go func() {
time.Sleep(60 * time.Millisecond)
mu.Unlock()
mu.Unlock()
mu.Unlock()
}()
time.Sleep(20 * time.Millisecond)
gtest.Assert(array.Len(), 1)
time.Sleep(50 * time.Millisecond)
gtest.Assert(array.Len(), 1)
time.Sleep(50 * time.Millisecond)
gtest.Assert(array.Len(), 3)
}