fix

src/server/config.min.ini

@@ -126,7 +126,7 @@
   partition_count = 4
 
 [pegasus.server]
-  perf_counter_enable_logging = false
+  perf_counter_enable_logging = true
   # Where the metrics are collected. If no value is given, no sink is used.
   # Options:
   #   - falcon

src/server/pegasus_read_service.h

@@ -43,13 +43,7 @@ class pegasus_read_service : public dsn::replication::replication_app_base,
     virtual ~pegasus_read_service() {}
     virtual int on_request(dsn::message_ex *request) override
     {
-        try {
-            handle_request(request);
-        } catch (TTransportException ex) {
-            derror_f("pegasus handle_request failed, from = {}, exception = {}",
-                     request->header->from_address.to_string(),
-                     ex.what());
-        }
+        handle_request(request);
         return 0;
     }
 

src/server/pegasus_server_write.cpp

@@ -33,6 +33,11 @@ namespace server {
 pegasus_server_write::pegasus_server_write(pegasus_server_impl *server, bool verbose_log)
     : replica_base(server), _write_svc(new pegasus_write_service(server)), _verbose_log(verbose_log)
 {
+    char name[256];
+    snprintf(name, 255, "recent_corrupt_write_count@%s", get_gpid().to_string());
+    _pfc_recent_corrupt_write_count.init_app_counter("app.pegasus",
+            name, COUNTER_TYPE_VOLATILE_NUMBER, "statistic the recent corrupt write count");
+
     init_non_batch_write_handlers();
 }
 
@@ -58,6 +63,7 @@ int pegasus_server_write::on_batched_write_requests(dsn::message_ex **requests,
             return iter->second(requests[0]);
         }
     } catch (TTransportException ex) {
+        _pfc_recent_corrupt_write_count->increment();
         derror_f("pegasus not batch write handler failed, from = {}, exception = {}",
                  requests[0]->header->from_address.to_string(),
                  ex.what());
@@ -101,6 +107,7 @@ int pegasus_server_write::on_batched_writes(dsn::message_ex **requests, int coun
                     }
                 }
             } catch (TTransportException ex) {
+                _pfc_recent_corrupt_write_count->increment();
                 derror_f("pegasus batch writes handler failed, from = {}, exception = {}",
                          requests[i]->header->from_address.to_string(),
                          ex.what());
@@ -111,12 +118,10 @@ int pegasus_server_write::on_batched_writes(dsn::message_ex **requests, int coun
             }
         }
 
-        if (dsn_likely(_put_rpc_batch.size() + _remove_rpc_batch.size() != 0)) {
-            if (err == 0) {
-                err = _write_svc->batch_commit(_decree);
-            } else {
-                _write_svc->batch_abort(_decree, err);
-            }
+        if (dsn_unlikely(err != 0 || _put_rpc_batch.size() + _remove_rpc_batch.size() == 0)) {
+            _write_svc->batch_abort(_decree, err == 0 ? -1 : err);
+        } else {
+            err = _write_svc->batch_commit(_decree);
         }
     }
 

src/server/pegasus_server_write.h

@@ -89,6 +89,8 @@ class pegasus_server_write : public dsn::replication::replica_base
 
     typedef std::map<dsn::task_code, std::function<int(dsn::message_ex *)>> non_batch_writes_map;
     non_batch_writes_map _non_batch_write_handlers;
+
+    ::dsn::perf_counter_wrapper _pfc_recent_corrupt_write_count;
 };
 
 } // namespace server

src/server/pegasus_write_service.h

@@ -173,7 +173,6 @@ class pegasus_write_service : dsn::replication::replica_base
 
     void set_default_ttl(uint32_t ttl);
 
-private:
     void clear_up_batch_states();
 
 private: