refactor: reimplement pegasus_server_write::on_batched_write_requests

[levy5307]

What problem does this PR solve?

reimplement pegasus_server_write::on_batched_write_requests

Check List

Tests

• Manual test

>>> ls
[general_info]
app_id  status     app_name  app_type  partition_count  replica_count  is_stateful  create_time          drop_time  drop_expire  envs_count  
1       AVAILABLE  stat      pegasus   4                3              true         2021-01-13_11:14:51  -          -            0           
2       AVAILABLE  temp      pegasus   8                3              true         2021-01-13_11:14:51  -          -            0           

[summary]
total_app_count  : 2

>>> use temp
OK
>>> multi_set a b1 c1 b2 c2
OK

app_id          : 2
partition_index : 4
decree          : 1
server          : 10.232.55.210:34802
>>> get a b1
"c1"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> get a b2
"c2"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> multi_del a b1 b2
2 key-value pairs deleted.

app_id          : 2
partition_index : 4
decree          : 3
server          : 10.232.55.210:34802
>>> get a b1
Not found

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> get a b2
Not found

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> set a b 1
OK

app_id          : 2
partition_index : 4
decree          : 5
server          : 10.232.55.210:34802
>>> incr a b 
2

app_id          : 2
partition_index : 4
decree          : 6
server          : 10.232.55.210:34802
>>> get a b
"2"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> check_and_set a -c b -t exist -s b3 -v c3 -r
hash_key: "a"
check_sort_key: "b"
check_type: exist
set_sort_key: "b3"
set_value: "c3"
set_value_ttl_seconds: 0
return_check_value: true

Set succeed.

Check value: "2"

app_id          : 2
partition_index : 4
decree          : 16
server          : 10.232.55.210:34802
>>> get a b3 c3
USAGE: 	get                     <hash_key> <sort_key>
>>> get a b3
"c3"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> check_and_mutate a -c b -t exist -r
Load mutations, like
  set <sort_key> <value> [ttl]
  del <sort_key>
Print "ok" to finish loading, "abort" to abort this command
>>> set b4 c4
LOAD: set sortkey "b4", value "c4", ttl 0
>>> set b5 c5
LOAD: set sortkey "b5", value "c5", ttl 0
>>> ok
INFO: load mutations done.

hash_key: "a"
check_sort_key: "b"
check_type: exist
return_check_value: true
mutations:
  mutation[0].type: SET
  mutation[0].sort_key: "b4"
  mutation[0].value: "c4"
  mutation[0].expire_seconds: 0
  mutation[1].type: SET
  mutation[1].sort_key: "b5"
  mutation[1].value: "c5"
  mutation[1].expire_seconds: 0

Mutate succeed.

Check value: "2"

app_id          : 2
partition_index : 4
decree          : 21
server          : 10.232.55.210:34802
>>> get a b4
"c4"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802
>>> get a b5
"c5"

app_id          : 2
partition_index : 4
server          : 10.232.55.210:34802